feat(qwp): expose drainer listener on the QuestDB facade and Sender builder

BackgroundDrainerListener was settable only on the internal drainer/pool
objects; users of the QuestDB pool entry point had no way to observe
background orphan-slot drainer events. Mirror the existing errorHandler/
connectionListener plumbing end to end:

- QuestDBBuilder.drainerListener(...) -> QuestDBImpl -> SenderPool
  applyUserCallbacks -> every pooled Sender (recovery delegates keep
  skipping user callbacks and forcing drain_orphans=off, unchanged)
- Sender.LineSenderBuilder.drainerListener(...) with the WebSocket-
  transport guard; build() installs it via setDrainerListener BEFORE
  startOrphanDrainers so drainers see the listener at submit time
- QwpWebSocketSender.setDrainerListener(...): synchronized with
  startOrphanDrainers; installs the pool submit-time default and
  re-assigns live drainers via pool.snapshot() when called late

E2E against the real TestWebSocketServer (new suppress-durable-ack-header
knob, setRejectWithStatus style): the facade-wired listener observes the
capability-gap episode (onDurableAckUnavailable 1..N) and the drain then
succeeds once the header is restored; a 421+REPLICA window is delivered on
onPrimaryUnavailable with onDurableAckUnavailable staying silent (M10
discrimination on the wire); same gap scenario one level down through
Sender.builder().drainerListener(...).

Author: bluestreak01

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

@@ -25,6 +25,7 @@
 package io.questdb.client;
 
 import io.questdb.client.cutlass.qwp.client.QwpQueryClient;
+import io.questdb.client.cutlass.qwp.client.sf.cursor.BackgroundDrainerListener;
 import io.questdb.client.impl.ConfigString;
 import io.questdb.client.impl.ConfigView;
 import io.questdb.client.impl.QuestDBImpl;
@@ -69,6 +70,7 @@ public final class QuestDBBuilder {
     // Optional ingest-side async callbacks. Null -> each pooled Sender uses its
     // loud-not-silent default. Applied to every Sender the pool builds.
     private SenderConnectionListener connectionListener;
+    private BackgroundDrainerListener drainerListener;
     private SenderErrorHandler errorHandler;
     private long housekeeperIntervalMillis = UNSET;
     private String config;
@@ -127,6 +129,25 @@ public QuestDBBuilder connectionListener(SenderConnectionListener listener) {
         return this;
     }
 
+    /**
+     * Sets the background orphan-slot drainer listener applied to every pooled
+     * ingest {@link Sender}. The listener observes the background drainer
+     * events of every sender the pool builds: durable-ack capability-gap
+     * retries, transient all-replica failover windows, and the eventual
+     * escalation to a {@code .failed} sentinel. Events are delivered on the
+     * drainers' own threads, so the listener must be thread-safe and must not
+     * block. Only meaningful when the configuration enables
+     * {@code drain_orphans}. Pass {@code null} (the default) to keep the
+     * drainers' default (no listener).
+     *
+     * @param listener the shared drainer listener, or {@code null} for the default
+     * @return this instance for method chaining
+     */
+    public QuestDBBuilder drainerListener(BackgroundDrainerListener listener) {
+        this.drainerListener = listener;
+        return this;
+    }
+
     /**
      * Sets the async error handler applied to every pooled ingest
      * {@link Sender}. The handler receives terminal/async ingest errors
@@ -209,7 +230,8 @@ public QuestDB build() {
                 housekeeperIntervalMillis,
                 queryCloseTimeoutMillis,
                 errorHandler,
-                connectionListener
+                connectionListener,
+                drainerListener
         );
     }
 

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

@@ -1022,6 +1022,11 @@ final class LineSenderBuilder {
         // Bounded inbox capacity for the async connection-event dispatcher.
         // PARAMETER_NOT_SET_EXPLICITLY → spec default (64).
         private int connectionListenerInboxCapacity = PARAMETER_NOT_SET_EXPLICITLY;
+        // Optional user-supplied observer for background orphan-slot drainer
+        // events (durable-ack capability-gap retries, all-replica failover
+        // windows, persistent-failure escalation). When null, drainers run
+        // without a listener. Only meaningful with drainOrphans=true.
+        private io.questdb.client.cutlass.qwp.client.sf.cursor.BackgroundDrainerListener drainerListener;
         // Orphan adoption: when true, the foreground sender scans
         // <sf_dir>/*/ at startup for sibling slots that hold unacked data
         // and reports them. Default false. Spec calls for spawning
@@ -1585,6 +1590,12 @@ public Sender build() {
                 // WebSocketClient inside the abandoned `connected`.
                 connected.setTransactional(transactional);
                 try {
+                    // Install the drainer listener BEFORE startOrphanDrainers
+                    // below: drainers must see the listener at submit time so
+                    // no early drainer event is lost to a late installation.
+                    if (drainerListener != null) {
+                        connected.setDrainerListener(drainerListener);
+                    }
                     // 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
@@ -1787,6 +1798,31 @@ public LineSenderBuilder disableAutoFlush() {
             return this;
         }
 
+        /**
+         * Sets the async listener observing background orphan-slot drainer
+         * events: per-attempt durable-ack capability-gap retries
+         * ({@code onDurableAckUnavailable}), transient all-replica failover
+         * windows ({@code onPrimaryUnavailable}), and the eventual escalation
+         * to a {@code .failed} sentinel
+         * ({@code onDurableAckPersistentFailure}). The listener runs on the
+         * drainers' own threads, so it must be thread-safe and must not block
+         * — hand off to a queue or metrics sink and return. Only meaningful
+         * when {@link #drainOrphans(boolean)} is enabled.
+         *
+         * <p>WebSocket transport only; setting on other transports throws.
+         *
+         * @param listener the listener; {@code null} keeps the default (no listener)
+         * @return this instance for method chaining
+         */
+        public LineSenderBuilder drainerListener(
+                io.questdb.client.cutlass.qwp.client.sf.cursor.BackgroundDrainerListener listener) {
+            if (protocol != PARAMETER_NOT_SET_EXPLICITLY && protocol != PROTOCOL_WEBSOCKET) {
+                throw new LineSenderException("drainer_listener is only supported for WebSocket transport");
+            }
+            this.drainerListener = listener;
+            return this;
+        }
+
         /**
          * Opt in to adopting sibling slots under {@code <sf_dir>/*} at
          * startup that hold unacked data left behind by a crashed sender or

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

@@ -39,6 +39,7 @@
 import io.questdb.client.cutlass.line.LineSenderException;
 import io.questdb.client.cutlass.line.array.DoubleArray;
 import io.questdb.client.cutlass.line.array.LongArray;
+import io.questdb.client.cutlass.qwp.client.sf.cursor.BackgroundDrainerListener;
 import io.questdb.client.cutlass.qwp.client.sf.cursor.BackgroundDrainerPool;
 import io.questdb.client.cutlass.qwp.client.sf.cursor.CursorSendEngine;
 import io.questdb.client.cutlass.qwp.client.sf.cursor.CursorWebSocketSendLoop;
@@ -205,6 +206,11 @@ public class QwpWebSocketSender implements Sender {
     private CursorSendEngine cursorEngine;
     private CursorWebSocketSendLoop cursorSendLoop;
     private boolean deferCommit;
+    // User-supplied observer for background orphan-slot drainer events.
+    // Volatile: written by setDrainerListener (any thread, before or after
+    // startOrphanDrainers) and read at pool-creation time. Null -> drainers
+    // run without a listener.
+    private volatile BackgroundDrainerListener drainerListener;
     // 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().
@@ -2079,6 +2085,33 @@ public void setCursorEngine(CursorSendEngine engine, boolean takeOwnership) {
         this.ownsCursorEngine = takeOwnership && engine != null;
     }
 
+    /**
+     * Register an async observer for background orphan-slot drainer events.
+     * May be called either before or after {@link #startOrphanDrainers} —
+     * when called before, the drainer pool picks it up as its submit-time
+     * default; when called after, it propagates to the pool AND to every
+     * live drainer (per-drainer re-assignment while running is explicitly
+     * permitted by the drainer's listener contract). Pass {@code null} to
+     * clear. {@code synchronized} to coordinate with
+     * {@code startOrphanDrainers}: a concurrent submit either observes the
+     * pool listener already set or is covered by the snapshot propagation.
+     */
+    public synchronized void setDrainerListener(BackgroundDrainerListener listener) {
+        this.drainerListener = listener;
+        BackgroundDrainerPool pool = drainerPool;
+        if (pool != null) {
+            // Submit-time fallback for drainers not yet submitted...
+            pool.setListener(listener);
+            // ...and direct re-assignment for the ones already running (the
+            // pool listener is only applied at submit time, never after).
+            ObjList<io.questdb.client.cutlass.qwp.client.sf.cursor.BackgroundDrainer> live =
+                    pool.snapshot();
+            for (int i = 0, n = live.size(); i < n; i++) {
+                live.getQuick(i).setListener(listener);
+            }
+        }
+    }
+
     /**
      * Configure the user-supplied error handler. May be called either before
      * or after {@code connect()} — when called after, the change propagates
@@ -2177,6 +2210,9 @@ public synchronized void startOrphanDrainers(
         if (drainerPool == null) {
             drainerPool = new io.questdb.client.cutlass.qwp.client.sf.cursor
                     .BackgroundDrainerPool(maxBackgroundDrainers);
+            // Install the user listener as the pool's submit-time default so
+            // the drainers submitted below observe it from their first event.
+            drainerPool.setListener(this.drainerListener);
         }
         for (int i = 0, n = orphanSlotPaths.size(); i < n; i++) {
             String slot = orphanSlotPaths.get(i);

core/src/main/java/io/questdb/client/impl/QuestDBImpl.java

@@ -30,6 +30,7 @@
 import io.questdb.client.SenderConnectionListener;
 import io.questdb.client.SenderErrorHandler;
 import io.questdb.client.cutlass.qwp.client.QwpQueryClient;
+import io.questdb.client.cutlass.qwp.client.sf.cursor.BackgroundDrainerListener;
 import org.jetbrains.annotations.TestOnly;
 
 import java.util.function.Consumer;
@@ -64,11 +65,13 @@ public QuestDBImpl(
             long housekeeperIntervalMillis,
             long queryCloseTimeoutMillis,
             SenderErrorHandler errorHandler,
-            SenderConnectionListener connectionListener
+            SenderConnectionListener connectionListener,
+            BackgroundDrainerListener drainerListener
     ) {
         this(ingestConfig, queryConfig, senderMin, senderMax, queryMin, queryMax,
                 acquireTimeoutMillis, idleTimeoutMillis, maxLifetimeMillis,
-                housekeeperIntervalMillis, queryCloseTimeoutMillis, null, null, errorHandler, connectionListener);
+                housekeeperIntervalMillis, queryCloseTimeoutMillis, null, null,
+                errorHandler, connectionListener, drainerListener);
     }
 
     // Test-only constructor exposing the senderFactory and connectHook seams:
@@ -95,13 +98,14 @@ public QuestDBImpl(
         this(ingestConfig, queryConfig, senderMin, senderMax, queryMin, queryMax,
                 acquireTimeoutMillis, idleTimeoutMillis, maxLifetimeMillis,
                 housekeeperIntervalMillis, QueryClientPool.DEFAULT_CLOSE_QUERY_TIMEOUT_MILLIS,
-                senderFactory, connectHook, null, null);
+                senderFactory, connectHook, null, null, null);
     }
 
-    // Full constructor adding the ingest-side errorHandler/connectionListener,
-    // applied by SenderPool to every Sender it builds. The 12-arg overload above
-    // is the unchanged white-box test seam and delegates here with null
-    // callbacks; the public overload delegates here with null test seams.
+    // Full constructor adding the ingest-side errorHandler/connectionListener/
+    // drainerListener, applied by SenderPool to every Sender it builds. The
+    // 12-arg overload above is the unchanged white-box test seam and delegates
+    // here with null callbacks; the public overload delegates here with null
+    // test seams.
     QuestDBImpl(
             String ingestConfig,
             String queryConfig,
@@ -117,7 +121,8 @@ public QuestDBImpl(
             IntFunction<Sender> senderFactory,
             Consumer<QwpQueryClient> connectHook,
             SenderErrorHandler errorHandler,
-            SenderConnectionListener connectionListener
+            SenderConnectionListener connectionListener,
+            BackgroundDrainerListener drainerListener
     ) {
         SenderPool builtSenderPool = null;
         QueryClientPool builtQueryPool = null;
@@ -130,7 +135,7 @@ public QuestDBImpl(
                     // build() never blocks on a slow / reachable-but-not-acking
                     // server; the housekeeper drives it via runStartupRecoveryStep().
                     true,
-                    errorHandler, connectionListener);
+                    errorHandler, connectionListener, drainerListener);
             builtQueryPool = new QueryClientPool(
                     queryConfig, queryMin, queryMax, acquireTimeoutMillis,
                     idleTimeoutMillis, maxLifetimeMillis, connectHook);

core/src/main/java/io/questdb/client/impl/SenderPool.java

@@ -29,6 +29,7 @@
 import io.questdb.client.SenderErrorHandler;
 import io.questdb.client.cutlass.line.LineSenderException;
 import io.questdb.client.cutlass.qwp.client.QwpWebSocketSender;
+import io.questdb.client.cutlass.qwp.client.sf.cursor.BackgroundDrainerListener;
 import io.questdb.client.cutlass.qwp.client.sf.cursor.OrphanScanner;
 import io.questdb.client.std.Files;
 import io.questdb.client.std.IntList;
@@ -102,6 +103,7 @@ public final class SenderPool implements AutoCloseable {
     // User-supplied ingest callbacks, shared across every pooled Sender this
     // pool builds. Null -> each sender keeps its loud-not-silent default.
     private final SenderConnectionListener connectionListener;
+    private final BackgroundDrainerListener drainerListener;
     private final SenderErrorHandler errorHandler;
     private final long idleTimeoutMillis;
     // Test seam. Production builds delegates via defaultSender(); white-box
@@ -195,7 +197,7 @@ public SenderPool(
             long maxLifetimeMillis
     ) {
         this(configurationString, minSize, maxSize, acquireTimeoutMillis,
-                idleTimeoutMillis, maxLifetimeMillis, null, false, null, null);
+                idleTimeoutMillis, maxLifetimeMillis, null, false, null, null, null);
     }
 
     // Test-only constructor exposing the senderFactory seam: production builds
@@ -239,14 +241,14 @@ public SenderPool(
     ) {
         this(configurationString, minSize, maxSize, acquireTimeoutMillis,
                 idleTimeoutMillis, maxLifetimeMillis, senderFactory,
-                deferStartupRecovery, null, null);
+                deferStartupRecovery, null, null, null);
     }
 
-    // Full constructor adding the user-supplied ingest callbacks (error handler
-    // and connection listener), applied to every Sender the pool builds (see
-    // buildManagedSlotSender). The public 6-arg ctor and the test-only
-    // senderFactory overloads above both delegate here with null callbacks; the
-    // pooled QuestDB handle calls this directly.
+    // Full constructor adding the user-supplied ingest callbacks (error
+    // handler, connection listener and background-drainer listener), applied
+    // to every Sender the pool builds (see buildManagedSlotSender). The public
+    // 6-arg ctor and the test-only senderFactory overloads above both delegate
+    // here with null callbacks; the pooled QuestDB handle calls this directly.
     SenderPool(
             String configurationString,
             int minSize,
@@ -257,13 +259,15 @@ public SenderPool(
             IntFunction<Sender> senderFactory,
             boolean deferStartupRecovery,
             SenderErrorHandler errorHandler,
-            SenderConnectionListener connectionListener
+            SenderConnectionListener connectionListener,
+            BackgroundDrainerListener drainerListener
     ) {
         if (minSize < 0 || maxSize < 1 || minSize > maxSize) {
             throw new IllegalArgumentException("invalid pool sizing: min=" + minSize + ", max=" + maxSize);
         }
         this.errorHandler = errorHandler;
         this.connectionListener = connectionListener;
+        this.drainerListener = drainerListener;
         this.senderFactory = senderFactory != null ? senderFactory : this::defaultSender;
         // An injected factory (tests) drives recovery too, preserving the
         // white-box recovery seam; production recovery forces OFF-mode connects
@@ -1048,6 +1052,9 @@ private Sender.LineSenderBuilder applyUserCallbacks(Sender.LineSenderBuilder bui
         if (connectionListener != null) {
             builder.connectionListener(connectionListener);
         }
+        if (drainerListener != null) {
+            builder.drainerListener(drainerListener);
+        }
         return builder;
     }