feat(ilp): background drainer pool — adopt orphan slots and replay them

Wires the drainer runtime onto the orphan-scanner foundation. With
drain_orphans=true the foreground sender now actually empties sibling
slots holding unacked data instead of just logging that they exist.

Per-drainer lifecycle:
1. Open CursorSendEngine on the slot — its constructor takes the slot
   lock; if another sender or drainer holds it, the engine throws and
   the drainer exits silently (LOCKED_BY_OTHER, not a failure).
2. Open a fresh WebSocketClient via the foreground sender's connect
   factory — separate connection, same auth/host/port/TLS config.
3. Run a CursorWebSocketSendLoop until ackedFsn catches up to the
   publishedFsn snapshot taken at startup.
4. On terminal failure (auth, recovery, budget), drop a .failed
   sentinel into the slot. Future scans skip it until an operator
   clears it manually — bounded retry, then human-in-the-loop.

Pool: bounded fixed-thread executor, daemon threads, sized by
max_background_drainers (default 4). Closes via cooperative stop +
3s grace; daemon threads ensure no JVM-exit blocking.

Visibility: QwpWebSocketSender#getBackgroundDrainers returns a snapshot
list of live drainers with {slot, target, acked, outcome, lastError}.

Test: ghost sender writes 30 distinct rows against a silent server and
closes fast — leaves an unacked slot. Foreground sender opens the same
group root with a different sender_id and drain_orphans=true against an
ack server; asserts every distinct payload reaches the new server. Plus
a sentinel-skip test confirming an operator-set .failed file disqualifies
the slot from the next foreground run's scan.

Empty active segments and stale hot spares are left in the slot dir per
spec decision #13 ("no automatic cleanup of empty slot dirs"); the
scanner's no-op behavior on empty slots makes this cheap.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: bluestreak01

core/src/main/java/io/questdb/client/Sender.java

@@ -1073,25 +1073,6 @@ public Sender build() {
                         }
                     }
                     slotPath = sfDir + "/" + senderId;
-                    // Orphan scan runs BEFORE we open our own slot — keeps
-                    // the scan's "exclude my slot" filter conceptually
-                    // simple. If the user opted in, log what we found so
-                    // they have visibility on pending drain candidates
-                    // until the drainer runtime lands.
-                    if (drainOrphans) {
-                        io.questdb.client.std.ObjList<String> orphans =
-                                io.questdb.client.cutlass.qwp.client.sf.cursor.OrphanScanner
-                                        .scan(sfDir, senderId);
-                        if (orphans.size() > 0) {
-                            org.slf4j.LoggerFactory.getLogger(LineSenderBuilder.class)
-                                    .info("found {} orphan slot(s) under {} (drainer "
-                                                    + "runtime not yet implemented; "
-                                                    + "max_background_drainers={}); "
-                                                    + "first paths: {}",
-                                            orphans.size(), sfDir, maxBackgroundDrainers,
-                                            sample(orphans, 3));
-                        }
-                    }
                 }
                 long actualSfAppendDeadlineNanos =
                         sfAppendDeadlineMillis == PARAMETER_NOT_SET_EXPLICITLY
@@ -1101,7 +1082,7 @@ public Sender build() {
                         slotPath, actualSfMaxBytes,
                         actualSfMaxTotalBytes, actualSfAppendDeadlineNanos);
                 try {
-                    return QwpWebSocketSender.connect(
+                    QwpWebSocketSender connected = QwpWebSocketSender.connect(
                             hosts.getQuick(0),
                             ports.getQuick(0),
                             wsTlsConfig,
@@ -1119,6 +1100,28 @@ public Sender build() {
                             actualReconnectMaxBackoffMillis,
                             initialConnectRetry
                     );
+                    // Once the foreground sender is up, dispatch drainers
+                    // for any sibling orphan slots. Scan AFTER we acquire
+                    // our own slot lock so we never accidentally try to
+                    // adopt our own data; the OrphanScanner.scan filter
+                    // also excludes our sender_id.
+                    if (drainOrphans && sfDir != null) {
+                        io.questdb.client.std.ObjList<String> orphans =
+                                io.questdb.client.cutlass.qwp.client.sf.cursor.OrphanScanner
+                                        .scan(sfDir, senderId);
+                        if (orphans.size() > 0) {
+                            org.slf4j.LoggerFactory.getLogger(LineSenderBuilder.class)
+                                    .info("dispatching drainers for {} orphan slot(s) under {} "
+                                                    + "(max_background_drainers={})",
+                                            orphans.size(), sfDir, maxBackgroundDrainers);
+                            connected.startOrphanDrainers(
+                                    orphans,
+                                    maxBackgroundDrainers,
+                                    actualSfMaxBytes,
+                                    actualSfMaxTotalBytes);
+                        }
+                    }
+                    return connected;
                 } catch (Throwable t) {
                     try {
                         cursorEngine.close();
@@ -1753,19 +1756,6 @@ public LineSenderBuilder senderId(String id) {
             return this;
         }
 
-        private static String sample(io.questdb.client.std.ObjList<String> list, int n) {
-            int take = Math.min(n, list.size());
-            StringBuilder sb = new StringBuilder("[");
-            for (int i = 0; i < take; i++) {
-                if (i > 0) sb.append(", ");
-                sb.append(list.get(i));
-            }
-            if (list.size() > take) {
-                sb.append(", ...(").append(list.size() - take).append(" more)");
-            }
-            return sb.append("]").toString();
-        }
-
         private static void validateSenderId(String id) {
             if (id == null || id.isEmpty()) {
                 throw new LineSenderException("sender_id must not be empty");

core/src/main/java/io/questdb/client/cutlass/qwp/client/QwpWebSocketSender.java

@@ -176,6 +176,11 @@ public class QwpWebSocketSender implements Sender {
     // true  → startup connect goes through the same retry-with-backoff
     //         loop as in-flight reconnect; auth failures still terminal.
     private boolean initialConnectRetry = false;
+    // Orphan-slot drainer pool. Non-null only when the builder requested
+    // drain_orphans=true AND we have a slot path to scan against. Closed
+    // alongside the cursor send loop in close().
+    private io.questdb.client.cutlass.qwp.client.sf.cursor.BackgroundDrainerPool
+            drainerPool;
     // Single volatile counter, single writer (the wire-side actor that
     // performs reconnect; for now: ensureConnected during recovery).
     // Bumped on every successful reconnect AND on initial recovery from
@@ -641,6 +646,17 @@ public void close() {
                     LOG.error("Error closing cursor send loop: {}", String.valueOf(e));
                 }
             }
+            // Drainer pool runs after the foreground I/O loop is wound
+            // down — drainers don't share state with the foreground, so
+            // ordering doesn't matter for correctness, just predictable
+            // shutdown.
+            if (drainerPool != null) {
+                try {
+                    drainerPool.close();
+                } catch (Throwable e) {
+                    LOG.error("Error closing drainer pool: {}", String.valueOf(e));
+                }
+            }
 
             // Always free resources the I/O thread never touches:
             // encoder and table buffers are user-thread-only.
@@ -985,6 +1001,52 @@ public long getTotalAcks() {
         return l == null ? 0L : l.getTotalAcks();
     }
 
+    /**
+     * Starts orphan drainers for the given list of slot paths. Each path
+     * gets its own drainer thread, capped at {@code maxBackgroundDrainers}
+     * concurrent. Drainers run until the slot is fully drained or a
+     * terminal error occurs (then they drop a {@code .failed} sentinel).
+     * <p>
+     * Should be called once, immediately after {@code connect()} returns.
+     * Subsequent calls add more drainers to the same pool.
+     */
+    public synchronized void startOrphanDrainers(
+            io.questdb.client.std.ObjList<String> orphanSlotPaths,
+            int maxBackgroundDrainers,
+            long segmentSizeBytes,
+            long sfMaxTotalBytes
+    ) {
+        if (orphanSlotPaths == null || orphanSlotPaths.size() == 0
+                || maxBackgroundDrainers <= 0) {
+            return;
+        }
+        if (drainerPool == null) {
+            drainerPool = new io.questdb.client.cutlass.qwp.client.sf.cursor
+                    .BackgroundDrainerPool(maxBackgroundDrainers);
+        }
+        for (int i = 0, n = orphanSlotPaths.size(); i < n; i++) {
+            String slot = orphanSlotPaths.get(i);
+            io.questdb.client.cutlass.qwp.client.sf.cursor.BackgroundDrainer drainer =
+                    new io.questdb.client.cutlass.qwp.client.sf.cursor.BackgroundDrainer(
+                            slot, segmentSizeBytes, sfMaxTotalBytes,
+                            this::buildAndConnect,
+                            reconnectMaxDurationMillis,
+                            reconnectInitialBackoffMillis,
+                            reconnectMaxBackoffMillis);
+            drainerPool.submit(drainer);
+        }
+    }
+
+    /**
+     * Snapshot of drainers the foreground sender has dispatched. Useful
+     * for monitoring orphan-drain progress without parsing logs.
+     */
+    public java.util.List<io.questdb.client.cutlass.qwp.client.sf.cursor.BackgroundDrainer>
+            getBackgroundDrainers() {
+        if (drainerPool == null) return java.util.Collections.emptyList();
+        return drainerPool.snapshot();
+    }
+
     /**
      * Frames re-sent on the post-reconnect catch-up window — i.e. frames
      * whose FSN was already on the wire before the drop. Useful for