fix(qwp): foreground-priority connect-walk lock; stop connect-phase drainers immediately on close (M11)

Three changes addressing M11 (close/reconnect latency under outage):

1. buildAndConnect no longer holds the sender monitor across network I/O.
   A dedicated ReentrantLock guards the connect walk (shared hostTracker
   round/health state, round counters, foreground lifecycle commits).
   Foreground walks use lock() and can only ever wait behind another
   foreground walk; background drainer walks use tryLock() and treat
   contention as a transport-shaped transient, deferring the sweep into
   their existing capped-backoff retry (Invariant B). The foreground's
   reconnect and close() paths can no longer queue behind a drainer's
   endpoint walk (previously up to endpoints x ~30s on an unfair monitor).

2. BackgroundDrainerPool.close() splits its stop policy: drainers that
   never started draining (still in connect-retry, ackedFsn < 0) are
   stop-signaled before the graceful window, exiting within ~50ms; only
   actively-draining drainers get the 2.5s grace. Cuts close() latency
   during an outage from ~3s to roughly one stop-check park chunk.

3. Documents the worst-case latency budgets: connect-walk lock hold
   formula on buildAndConnect, close() budget on QwpWebSocketSender, and
   a close-latency note on the drainOrphans builder javadoc.

Author: bluestreak01

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

@@ -1840,6 +1840,16 @@ public LineSenderBuilder drainerListener(
          * Slots flagged with the {@code .failed} sentinel are skipped
          * (manual reset required), and the foreground sender's own slot is
          * never adopted.
+         * <p>
+         * Close-latency note: {@code close()} stops adopted drainers. A
+         * drainer still connecting (e.g. during an outage) is stop-signaled
+         * immediately and exits within ~50ms; a drainer actively replaying
+         * frames is given a ~2.5s grace window to finish, plus a 0.5s stop
+         * window — so {@code close()} may take up to ~3s while orphan
+         * drainers are in flight (and a drainer parked in a blocking native
+         * connect is abandoned to exit on its own daemon thread).
+         * Un-drained slots stay on disk and are re-adopted by the next
+         * sender that enables {@code drain_orphans}.
          */
         public LineSenderBuilder drainOrphans(boolean enabled) {
             if (protocol != PARAMETER_NOT_SET_EXPLICITLY && protocol != PROTOCOL_WEBSOCKET) {

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

@@ -73,6 +73,7 @@
 import java.util.List;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicReference;
+import java.util.concurrent.locks.ReentrantLock;
 
 /**
  * QWP v1 WebSocket client sender for streaming data to QuestDB.
@@ -152,6 +153,16 @@ public class QwpWebSocketSender implements Sender {
     private final List<Endpoint> endpoints;
     // Global symbol dictionary for delta encoding
     private final GlobalSymbolDictionary globalSymbolDictionary;
+    // Guards the connect walk (see buildAndConnect): the shared
+    // QwpHostHealthTracker round/health state, roundSeq,
+    // roundConnectAttemptSeq, and the foreground lifecycle commits
+    // (currentEndpointIdx, hasEverConnected, cap-derived sizing). A
+    // dedicated lock rather than the sender monitor so that (a) unrelated
+    // synchronized sender methods (setDrainerListener, startOrphanDrainers)
+    // never contend with a connect walk's network I/O, and (b) background
+    // walks can use tryLock to yield to the foreground instead of queuing
+    // on an unfair monitor.
+    private final ReentrantLock connectWalkLock = new ReentrantLock();
     private final QwpHostHealthTracker hostTracker;
     private final CharSequenceObjHashMap<QwpTableBuffer> tableBuffers;
     // null means plain text (no TLS)
@@ -943,6 +954,30 @@ public QwpWebSocketSender charColumn(CharSequence columnName, char value) {
         return this;
     }
 
+    /**
+     * Closes the sender: flushes user-thread state into the engine, drains
+     * acked data within {@code close_flush_timeout}, stops the I/O loop,
+     * closes the orphan-drainer pool, and frees buffers.
+     * <p>
+     * Worst-case latency budget (dominant contributors, sequential):
+     * <ul>
+     *   <li>bounded drain: up to {@code close_flush_timeout} when the server
+     *       is slow or unreachable ({@code <= 0} opts out);</li>
+     *   <li>I/O loop stop: the shutdown-latch await is untimed, but the loop
+     *       exits promptly unless the I/O thread sits inside a blocking
+     *       native connect — bounded by {@code connect_timeout}, or by the
+     *       OS SYN-retry deadline (60-130s on Linux) when the default
+     *       {@code 0} is in effect. Background drainer walks never delay
+     *       this stop: they yield the connect-walk lock to the foreground
+     *       (see {@link #buildAndConnect});</li>
+     *   <li>drainer pool: drainers still in their connect-retry phase are
+     *       stop-signaled immediately (exit within ~50ms); drainers actively
+     *       replaying frames get a 2.5s grace window plus a 0.5s stop window
+     *       — worst case ~3s when a drainer sits in a blocking native
+     *       connect (15s background deadline) and must be abandoned to exit
+     *       on its own.</li>
+     * </ul>
+     */
     @Override
     public void close() {
         if (!closed) {
@@ -2524,7 +2559,62 @@ private WebSocketClient newWebSocketClient() {
                 : WebSocketClientFactory.newPlainTextInstance();
     }
 
-    private synchronized WebSocketClient buildAndConnect(ReconnectSupplier ctx) {
+    /**
+     * Multi-endpoint connect walk shared by the foreground sender and the
+     * background orphan drainers. One invocation walks every endpoint of the
+     * current round, performing a TCP/TLS connect plus a WebSocket upgrade
+     * per endpoint, all while holding {@link #connectWalkLock} (the lock
+     * protects the shared {@link QwpHostHealthTracker} round/health state,
+     * the round counters, and the foreground lifecycle commits).
+     * <p>
+     * Worst-case lock hold per walk is
+     * {@code endpoints x (connect timeout + upgrade timeout)}:
+     * <ul>
+     *   <li>foreground walk: {@code connect_timeout} verbatim -- the default
+     *       {@code 0} keeps the untimed native connect, bounded only by the
+     *       OS SYN-retry deadline (60-130s per endpoint on Linux) -- plus
+     *       {@code auth_timeout_ms} (default 15s) for the upgrade;</li>
+     *   <li>background walk: 15s connect fallback
+     *       ({@link #DEFAULT_BACKGROUND_CONNECT_TIMEOUT_MS}) plus
+     *       {@code auth_timeout_ms} -- see
+     *       {@link #effectiveConnectTimeoutMs(boolean, int)}.</li>
+     * </ul>
+     * <p>
+     * Lock policy -- the foreground has absolute priority. Foreground walks
+     * (the producer's initial connect and the I/O loop's reconnects) block
+     * on {@code lock()} and can therefore only ever wait behind another
+     * foreground walk. Background walks use {@code tryLock()} and treat
+     * contention as a transport-shaped transient: the thrown
+     * {@link LineSenderException} lands in the drainer's indefinite
+     * capped-backoff retry (Invariant B) -- both in
+     * {@code BackgroundDrainer.connectWithDurableAckRetry()} and in
+     * {@code CursorWebSocketSendLoop.connectLoop}'s Throwable branch -- so a
+     * drainer sweep is deferred, never failed, and the foreground's
+     * reconnect and {@code close()} paths are never queued behind a
+     * drainer's endpoint walk.
+     */
+    private WebSocketClient buildAndConnect(ReconnectSupplier ctx) {
+        if (ctx.isBackground()) {
+            if (!connectWalkLock.tryLock()) {
+                // Transport-shaped on purpose: every drainer retry loop
+                // treats an untyped connect failure as transient and backs
+                // off (Invariant B) -- the sweep is deferred, never failed,
+                // and no .failed sentinel can result from lock contention.
+                throw new LineSenderException(
+                        "connect walk lock is busy (another connect walk is in progress); "
+                                + "background drainer will retry after backoff");
+            }
+        } else {
+            connectWalkLock.lock();
+        }
+        try {
+            return connectWalk(ctx);
+        } finally {
+            connectWalkLock.unlock();
+        }
+    }
+
+    private WebSocketClient connectWalk(ReconnectSupplier ctx) {
         // Background (drainer) factories share this connect walk -- endpoint
         // list, hostTracker health and round state -- but must stay INVISIBLE
         // in the foreground sender's observable state. SenderConnectionEvents

core/src/main/java/io/questdb/client/cutlass/qwp/client/sf/cursor/BackgroundDrainerPool.java

@@ -46,10 +46,15 @@
  * (no orphans submitted) costs one core thread; submitted-and-finished
  * drainers are GC'd after they complete.
  * <p>
- * Closing the pool requests every still-running drainer to stop and
- * waits up to a few seconds for them to exit cleanly. Drainers that
- * don't exit in time are left to finish on their own — the pool's
- * underlying executor uses daemon threads so they don't block JVM exit.
+ * Closing the pool uses a split stop policy: drainers that never started
+ * draining (still inside their connect-retry loop — e.g. the cluster is
+ * unreachable) are stop-signaled immediately, because no grace window can
+ * help them finish; drainers actively replaying frames get a graceful
+ * window to reach {@code acked >= target} before being signaled. Drainers
+ * that don't exit in time (typically parked in a blocking native connect
+ * that neither unpark nor interrupt cancels) are left to finish on their
+ * own — the pool's underlying executor uses daemon threads so they don't
+ * block JVM exit.
  */
 public final class BackgroundDrainerPool implements QuietCloseable {
 
@@ -66,9 +71,12 @@ public final class BackgroundDrainerPool implements QuietCloseable {
     // either lands before close (and close waits for it to finish) or
     // sees the closed bit and throws.
     private static final int CLOSED_BIT = Integer.MIN_VALUE;
-    // Time we let drainers finish their drain naturally before signaling
-    // stop. awaitTermination returns as soon as the last drainer exits,
-    // so this only matters when something is genuinely stuck.
+    // Time we let ACTIVELY DRAINING drainers finish naturally before
+    // signaling stop. Connect-phase drainers are stop-signaled before this
+    // window even starts (see close()), so during an outage — when no
+    // drainer can be draining — close() does not pay this in full.
+    // awaitTermination returns as soon as the last drainer exits, so this
+    // only matters when something is genuinely stuck.
     private static final long GRACEFUL_DRAIN_MILLIS = 2_500L;
     private static final Logger LOG = LoggerFactory.getLogger(BackgroundDrainerPool.class);
     // After signaling stop, give drainers a brief window to unwind cleanly
@@ -125,11 +133,33 @@ public void close() {
         while (state.get() != CLOSED_BIT) {
             Compat.onSpinWait();
         }
-        // Reject new tasks but let in-flight drainers finish their drain
-        // naturally. Without this grace window a drainer that's seconds
-        // away from acked >= target gets requestStop()'d and exits as
-        // STOPPED — its engine.close() then sees fullyDrained=false and
-        // leaves the slot's .sfa files behind, defeating drain_orphans.
+        // Split stop policy. The graceful window below exists so a drainer
+        // that is seconds away from acked >= target is not cut down
+        // mid-drain (its engine.close() would see fullyDrained=false and
+        // leave the slot's .sfa files behind, defeating drain_orphans). A
+        // drainer that never started draining — still inside its
+        // connect-retry loop, e.g. the cluster is unreachable and
+        // Invariant B retries forever — cannot possibly use that window
+        // productively, so stop it NOW: it wakes from its backoff park
+        // within ~50ms (STOP_CHECK_PARK_CHUNK_NANOS) and exits as STOPPED,
+        // cutting close() latency during an outage from
+        // GRACEFUL_DRAIN_MILLIS + STOP_GRACE_MILLIS (~3s) to roughly one
+        // stop-check park chunk. ackedFsn stays -1 until the drain loop's
+        // first poll, so `< 0` discriminates "never connected/started
+        // draining" from "actively draining"; the moments-wide race with a
+        // just-connected drainer is benign — it exits as STOPPED and the
+        // slot is re-adopted by the next scan. A drainer parked inside a
+        // blocking native connect ignores the stop until its background
+        // connect deadline resolves; that one still burns the full grace +
+        // stop windows below and is then abandoned to exit on its own
+        // (daemon thread).
+        for (BackgroundDrainer d : active) {
+            if (d.outcome() == BackgroundDrainer.DrainOutcome.PENDING && d.getAckedFsn() < 0) {
+                d.requestStop();
+            }
+        }
+        // Reject new tasks but let actively-draining drainers finish
+        // naturally.
         executor.shutdown();
         try {
             if (!executor.awaitTermination(GRACEFUL_DRAIN_MILLIS, TimeUnit.MILLISECONDS)) {

core/src/test/java/io/questdb/client/test/cutlass/qwp/client/QwpWebSocketSenderJvmErrorCleanupTest.java

@@ -151,11 +151,15 @@ public void testExceptionPathStillClosesAndWalksAllEndpoints() throws Exception
      * Bypasses the real constructor -- no wire client, engine or dispatcher
      * needed. The connect walk dereferences only the fields wired below plus
      * primitives whose zero-defaults are valid here (field initializers do
-     * not run under {@code Unsafe.allocateInstance}).
+     * not run under {@code Unsafe.allocateInstance}), plus the connect-walk
+     * lock, which buildAndConnect acquires unconditionally and is therefore
+     * wired here.
      */
     private static QwpWebSocketSender newBareSender() throws Exception {
-        return (QwpWebSocketSender) Unsafe.getUnsafe()
+        QwpWebSocketSender sender = (QwpWebSocketSender) Unsafe.getUnsafe()
                 .allocateInstance(QwpWebSocketSender.class);
+        setField(sender, "connectWalkLock", new java.util.concurrent.locks.ReentrantLock());
+        return sender;
     }
 
     private static QwpHostHealthTracker wireEndpoints(QwpWebSocketSender sender,