refactor(qwp): lock-free background connect walks via private round cursors (M11 follow-up)

Completes the M11 concurrency work: no network I/O runs under a
sender-wide lock for background work.

QwpHostHealthTracker (additive):
- newRoundCursor(): walker-private full-sweep cursor -- claim-at-pick
  over cursor-local attempted bits, ordered by the live shared
  (state, zone_tier) priority. Makes the pick->record pair safe without
  external serialization and guarantees every sweep tries every endpoint
  exactly once regardless of concurrent walkers.
- Health-only record overloads (markRoundAttempted=false) for
  recordSuccess/recordTransportError/recordRoleReject: update the shared
  health ledger without touching the shared round's attempted bits.
  Existing signatures delegate with true; the query client is untouched.

QwpWebSocketSender:
- Background (drainer) walks take NO lock: private RoundCursor +
  health-only records. Concurrent drainer sweeps proceed in parallel
  with each other and with the foreground, share health observations,
  and can neither consume nor poison the foreground's round (roundSeq
  and the round lifecycle are now strictly foreground-owned).
- The tryLock-yield path (and its busy exception) is gone: background
  sweeps are no longer deferred on contention, they just run.
- Foreground walks keep connectWalkLock across the sweep -- they own the
  shared round state and cannot overlap by construction; the lock is
  insurance for that invariant.

Tests:
- QwpHostHealthTrackerTest: +6 tests -- cursor full-sweep priority
  order, live re-ranking mid-sweep, shared-round isolation, no endpoint
  stealing between cursors, health-only records vs round bits,
  health-only success feeding sticky-Healthy recency.
- QwpConnectWalkLockPriorityTest replaced by
  QwpConnectWalkBackgroundIsolationTest: two full background sweeps run
  to completion while a foreground walk is parked inside a blocking
  connect holding the lock. Red-verified against the previous
  tryLock design: fails with the busy-exception artifact.
- BackgroundDrainerPoolConnectPhaseCloseTest: comments updated (the
  busy exception no longer exists; the factory now throws a generic
  outage-shaped transport error -- same Invariant B contract).

Author: bluestreak01

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

@@ -35,9 +35,19 @@
  * so a known-good cross-zone host is picked before an untried local host.
  * <p>
  * Each method is internally synchronized, but pickNext + recordX is not atomic
- * across the pair. Callers must externally serialize a pick → record sequence
- * (the QWP clients do this via the sender's {@code synchronized buildAndConnect}
- * and the query client's documented one-execute-at-a-time contract).
+ * across the pair. Callers of the SHARED-round API (pickNext / beginRound /
+ * isRoundExhausted) must externally serialize a pick → record sequence (the
+ * ingest sender does this by keeping its foreground connect walk single-file
+ * behind a lock; the query client via its documented one-execute-at-a-time
+ * contract).
+ * <p>
+ * Concurrent walkers that must not consume or poison the shared round --
+ * the ingest sender's background orphan drainers -- use a private
+ * {@link RoundCursor} ({@link #newRoundCursor()}) paired with the
+ * health-only record overloads ({@code markRoundAttempted=false}): the
+ * cursor's attempted set is walker-local (claim-at-pick, so concurrent
+ * cursors never race on the pick → record pair), while state/zone updates
+ * flow into the shared health ledger that orders everyone's picks.
  */
 public final class QwpHostHealthTracker {
     public enum HostState {
@@ -250,24 +260,113 @@ public void recordMidStreamFailure(int idx) {
     }
 
     public void recordRoleReject(int idx, boolean isTransient) {
+        recordRoleReject(idx, isTransient, true);
+    }
+
+    /**
+     * Variant with an explicit round-bit policy. {@code markRoundAttempted =
+     * false} updates only the shared health ledger (state), leaving the
+     * shared round's attempted bit untouched — for walkers on a private
+     * {@link RoundCursor} whose attempts must stay invisible to the shared
+     * round (the ingest sender's background drainers).
+     */
+    public void recordRoleReject(int idx, boolean isTransient, boolean markRoundAttempted) {
         synchronized (lock) {
             states[idx] = isTransient ? HostState.TRANSIENT_REJECT : HostState.TOPOLOGY_REJECT;
-            attemptedThisRound[idx] = true;
+            if (markRoundAttempted) {
+                attemptedThisRound[idx] = true;
+            }
         }
     }
 
     public void recordSuccess(int idx) {
+        recordSuccess(idx, true);
+    }
+
+    /**
+     * Variant with an explicit round-bit policy; see
+     * {@link #recordRoleReject(int, boolean, boolean)}. The success epoch
+     * (sticky-Healthy recency) is recorded either way — a background
+     * walker's success is real health data.
+     */
+    public void recordSuccess(int idx, boolean markRoundAttempted) {
         synchronized (lock) {
             states[idx] = HostState.HEALTHY;
-            attemptedThisRound[idx] = true;
+            if (markRoundAttempted) {
+                attemptedThisRound[idx] = true;
+            }
             lastSuccessEpoch[idx] = ++successEpoch;
         }
     }
 
     public void recordTransportError(int idx) {
+        recordTransportError(idx, true);
+    }
+
+    /**
+     * Variant with an explicit round-bit policy; see
+     * {@link #recordRoleReject(int, boolean, boolean)}.
+     */
+    public void recordTransportError(int idx, boolean markRoundAttempted) {
         synchronized (lock) {
             states[idx] = HostState.TRANSPORT_ERROR;
-            attemptedThisRound[idx] = true;
+            if (markRoundAttempted) {
+                attemptedThisRound[idx] = true;
+            }
+        }
+    }
+
+    /**
+     * Creates a walker-private full-sweep cursor over the host list. Each
+     * {@link RoundCursor#next()} returns the highest-priority host this
+     * cursor has not yet returned — priority is the same live
+     * {@code (state, zone_tier)} tuple {@link #pickNext()} uses — and
+     * claims it at pick time in the cursor's OWN attempted set, so:
+     * <ul>
+     *   <li>every cursor sweeps every host exactly once regardless of what
+     *       other walkers do concurrently (no endpoint stealing);</li>
+     *   <li>the pick → record pair needs no external serialization — the
+     *       claim is cursor-local, and the health records are atomic;</li>
+     *   <li>the shared round (attempted bits, {@link #beginRound},
+     *       {@link #isRoundExhausted}) is never consulted nor mutated.</li>
+     * </ul>
+     * Pair with the {@code markRoundAttempted=false} record overloads so the
+     * walker's results update shared health without touching the shared
+     * round.
+     */
+    public RoundCursor newRoundCursor() {
+        return new RoundCursor();
+    }
+
+    /** See {@link #newRoundCursor()}. Not thread-safe for sharing a single
+     *  instance across walkers; create one per walk. */
+    public final class RoundCursor {
+        private final boolean[] attempted = new boolean[hostCount];
+
+        private RoundCursor() {
+        }
+
+        /**
+         * Highest-priority host this cursor has not yet returned, claimed at
+         * pick time; -1 once the cursor has swept every host. Ordering reads
+         * the LIVE shared health state under the tracker lock, so a state
+         * change recorded by any walker between two calls re-ranks the
+         * remaining hosts.
+         */
+        public int next() {
+            synchronized (lock) {
+                for (HostState p : PRIORITY_ORDER) {
+                    for (ZoneTier z : ZONE_PRIORITY_ORDER) {
+                        for (int i = 0; i < hostCount; i++) {
+                            if (!attempted[i] && states[i] == p && zoneTiers[i] == z) {
+                                attempted[i] = true;
+                                return i;
+                            }
+                        }
+                    }
+                }
+                return -1;
+            }
         }
     }
 

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

@@ -153,15 +153,19 @@ 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.
+    // Serializes FOREGROUND connect walks only (see buildAndConnect): the
+    // shared-round state in hostTracker (pickNext/beginRound/attempted
+    // bits), roundSeq, roundConnectAttemptSeq, and the foreground lifecycle
+    // commits (currentEndpointIdx, hasEverConnected, cap-derived sizing)
+    // all have exactly one writer -- the foreground walk -- and foreground
+    // walks cannot overlap by construction (the I/O loop is single-threaded
+    // and the user-thread initial connect completes before the loop
+    // starts); the lock is cheap insurance for that invariant. Background
+    // (drainer) walks take NO lock at all: they walk a private
+    // QwpHostHealthTracker.RoundCursor and record health-only results, so
+    // no network I/O ever runs under a sender-wide lock for background
+    // work, and neither the foreground's reconnect nor close() can queue
+    // behind a drainer's endpoint walk.
     private final ReentrantLock connectWalkLock = new ReentrantLock();
     private final QwpHostHealthTracker hostTracker;
     private final CharSequenceObjHashMap<QwpTableBuffer> tableBuffers;
@@ -968,8 +972,9 @@ public QwpWebSocketSender charColumn(CharSequence columnName, char value) {
      *       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>
+     *       this stop: they run lock-free on private round cursors and
+     *       never hold anything the foreground waits on (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
@@ -2585,13 +2590,9 @@ private WebSocketClient newWebSocketClient() {
 
     /**
      * 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
+     * background orphan drainers. One invocation sweeps the endpoint list,
+     * performing a TCP/TLS connect plus a WebSocket upgrade per endpoint;
+     * worst-case sweep duration is
      * {@code endpoints x (connect timeout + upgrade timeout)}:
      * <ul>
      *   <li>foreground walk: {@code connect_timeout} verbatim -- the default
@@ -2604,33 +2605,30 @@ private WebSocketClient newWebSocketClient() {
      *       {@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.
+     * Concurrency policy -- no network I/O under a sender-wide lock for
+     * background work. FOREGROUND walks (the producer's initial connect and
+     * the I/O loop's reconnects) hold {@link #connectWalkLock} across the
+     * sweep: they own the shared round state and the lifecycle commits, and
+     * can only ever wait behind another foreground walk (which cannot
+     * happen by construction -- the lock is insurance). BACKGROUND (drainer)
+     * walks take NO lock: each sweeps a private
+     * {@link QwpHostHealthTracker.RoundCursor} -- full sweep, claim-at-pick,
+     * ordered by the live shared health state -- and records results with
+     * the health-only overloads ({@code markRoundAttempted=false}), so
+     * concurrent drainer sweeps proceed in parallel with each other and
+     * with the foreground, share health observations, and can neither
+     * consume nor poison the foreground's round. The foreground's
+     * reconnect and {@code close()} paths are therefore 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();
+            // Lock-free: the walk below touches only internally-synchronized
+            // hostTracker health state and walk-local/cursor-local state on
+            // the background path.
+            return connectWalk(ctx);
         }
+        connectWalkLock.lock();
         try {
             return connectWalk(ctx);
         } finally {
@@ -2640,7 +2638,10 @@ private WebSocketClient buildAndConnect(ReconnectSupplier ctx) {
 
     private WebSocketClient connectWalk(ReconnectSupplier ctx) {
         // Background (drainer) factories share this connect walk -- endpoint
-        // list, hostTracker health and round state -- but must stay INVISIBLE
+        // list and hostTracker HEALTH state (never the shared round: a
+        // background sweep walks its own RoundCursor and records with
+        // markRoundAttempted=false, so it cannot consume the foreground's
+        // round or skew roundSeq) -- but must stay INVISIBLE
         // in the foreground sender's observable state. SenderConnectionEvents
         // describe the FOREGROUND connection's lifecycle, and the cap-derived
         // sizing (serverMaxBatchSize / effectiveAutoFlushBytes) guards the
@@ -2651,6 +2652,12 @@ private WebSocketClient connectWalk(ReconnectSupplier ctx) {
         // (oversize batch -> ws-close[1009] -> producer-terminal HALT caused
         // by background activity).
         final boolean background = ctx.isBackground();
+        // Private full-sweep cursor for background walks: claim-at-pick over
+        // cursor-local attempted bits makes the pick -> record pair safe
+        // without any walk-wide lock, and guarantees every sweep tries every
+        // endpoint exactly once regardless of concurrent walkers.
+        final QwpHostHealthTracker.RoundCursor cursor =
+                background ? hostTracker.newRoundCursor() : null;
         int previousIdx = ctx.previousIdx;
         if (previousIdx >= 0) {
             // Mid-stream wire failure -- the I/O loop just observed the active
@@ -2673,7 +2680,10 @@ private WebSocketClient connectWalk(ReconnectSupplier ctx) {
             hostTracker.recordMidStreamFailure(previousIdx);
             ctx.previousIdx = -1;
         }
-        if (hostTracker.isRoundExhausted()) {
+        // Shared-round lifecycle is foreground-only: a background walk must
+        // not advance the round (or roundSeq, which numbers foreground
+        // events) under the foreground's feet.
+        if (!background && hostTracker.isRoundExhausted()) {
             roundSeq++;
             hostTracker.beginRound(true);
         }
@@ -2693,7 +2703,7 @@ private WebSocketClient connectWalk(ReconnectSupplier ctx) {
             if (ctx.isAborted()) {
                 throw new LineSenderException(ctx.abortMessage());
             }
-            int idx = hostTracker.pickNext();
+            int idx = background ? cursor.next() : hostTracker.pickNext();
             if (idx < 0) break;
             Endpoint ep = endpoints.get(idx);
             lastEndpoint = ep;
@@ -2717,7 +2727,7 @@ private WebSocketClient connectWalk(ReconnectSupplier ctx) {
                 newClient.close();
                 if (classified instanceof QwpIngressRoleRejectedException) {
                     QwpIngressRoleRejectedException re = (QwpIngressRoleRejectedException) classified;
-                    hostTracker.recordRoleReject(idx, re.isTransient());
+                    hostTracker.recordRoleReject(idx, re.isTransient(), !background);
                     lastError = re;
                     lastRoleReject = re;
                     if (!background) {
@@ -2749,7 +2759,7 @@ private WebSocketClient connectWalk(ReconnectSupplier ctx) {
                                 && !((WebSocketUpgradeException) classified).isRoleMismatch()))) {
                     terminalUpgradeError = classified;
                 }
-                hostTracker.recordTransportError(idx);
+                hostTracker.recordTransportError(idx, !background);
                 lastError = classified;
                 if (!background) {
                     dispatchConnectionEvent(
@@ -2760,7 +2770,7 @@ private WebSocketClient connectWalk(ReconnectSupplier ctx) {
                 continue;
             } catch (Exception e) {
                 newClient.close();
-                hostTracker.recordTransportError(idx);
+                hostTracker.recordTransportError(idx, !background);
                 lastError = e;
                 if (!background) {
                     dispatchConnectionEvent(
@@ -2790,7 +2800,7 @@ private WebSocketClient connectWalk(ReconnectSupplier ctx) {
             }
             if (requestDurableAck && !newClient.isServerDurableAckEnabled()) {
                 newClient.close();
-                hostTracker.recordRoleReject(idx, false);
+                hostTracker.recordRoleReject(idx, false, !background);
                 QwpDurableAckMismatchException ackErr = new QwpDurableAckMismatchException(
                         ep.host, ep.port, null);
                 if (terminalUpgradeError == null) {
@@ -2805,7 +2815,7 @@ private WebSocketClient connectWalk(ReconnectSupplier ctx) {
                 }
                 continue;
             }
-            hostTracker.recordSuccess(idx);
+            hostTracker.recordSuccess(idx, !background);
             ctx.previousIdx = idx;
             if (background) {
                 // Walk bookkeeping only: recordSuccess feeds the shared health