fix(qwp): mid-drain durable-ack capability gap re-enters the settle budget instead of quarantining on the first sweep

A capability gap surfacing through a mid-drain reconnect sweep used to
hit BackgroundDrainer.run()'s generic wire-error path and quarantine the
slot immediately, bypassing the 16-attempt/wall-clock settle budget that
the initial-connect path grants the exact same condition (rolling
upgrade: primary restarts as a non-durable-ack build while a drain is in
flight).

CursorWebSocketSendLoop now latches a typed capabilityGapTerminal marker
(written before the terminalError volatile, first-writer-wins) that the
drainer consults after checkError() throws: on a gap terminal it
recycles loop + client and re-enters connectWithDurableAckRetry(), which
owns the episode budget and drops the .failed sentinel itself if the gap
persists. Every other terminal keeps the original quarantine path, and
the foreground sender keeps its spec'd loud-fail (the marker is
package-private and never consulted there).

Author: bluestreak01

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

@@ -159,7 +159,10 @@ public BackgroundDrainer() {
     }
 
     /**
-     * Initial connect with retry on whole-cluster durable-ack unavailability.
+     * Budgeted connect with retry on whole-cluster durable-ack unavailability:
+     * the initial connect, and re-entered from {@link #run()} whenever a
+     * mid-drain reconnect sweep hits the same capability gap (each re-entry
+     * is a fresh episode -- a successful connect ended the previous one).
      * The wrapped {@code clientFactory.reconnect()} already walks every
      * configured endpoint per attempt and only throws
      * {@link QwpDurableAckMismatchException} when none of them advertise
@@ -429,37 +432,78 @@ public void run() {
                 // already dropped on the FAILED path.
                 return;
             }
-            loop = new CursorWebSocketSendLoop(
-                    client, engine,
-                    0L, CursorWebSocketSendLoop.DEFAULT_PARK_NANOS,
-                    clientFactory,
-                    reconnectMaxDurationMillis,
-                    reconnectInitialBackoffMillis,
-                    reconnectMaxBackoffMillis,
-                    requestDurableAck,
-                    durableAckKeepaliveIntervalMillis);
-            loop.start();
-
+            // One iteration per wire session. Re-entered ONLY when a mid-drain
+            // reconnect sweep hit a durable-ack CAPABILITY gap: that is the
+            // exact rolling-upgrade condition the settle budget in
+            // connectWithDurableAckRetry() exists for, so it must not
+            // quarantine on the first sweep the way the initial-connect path
+            // never does. The engine stays alive across sessions (it holds the
+            // slot lock; only loop + client are recycled), and target remains
+            // valid -- the slot is orphaned, nothing appends to it.
+            drain:
             while (!stopRequested) {
-                long acked = engine.ackedFsn();
-                this.ackedFsn = acked;
-                if (acked >= target) {
-                    outcome = DrainOutcome.SUCCESS;
-                    LOG.info("drainer fully drained slot {} (target={}, acked={})",
-                            slotPath, target, acked);
-                    return;
-                }
-                try {
-                    loop.checkError();
-                } catch (Throwable t) {
-                    String msg = t.getMessage();
-                    LOG.error("drainer wire error for slot {}: {}", slotPath, msg);
-                    lastErrorMessage = msg;
-                    OrphanScanner.markFailed(slotPath, "wire: " + msg);
-                    outcome = DrainOutcome.FAILED;
-                    return;
+                loop = new CursorWebSocketSendLoop(
+                        client, engine,
+                        0L, CursorWebSocketSendLoop.DEFAULT_PARK_NANOS,
+                        clientFactory,
+                        reconnectMaxDurationMillis,
+                        reconnectInitialBackoffMillis,
+                        reconnectMaxBackoffMillis,
+                        requestDurableAck,
+                        durableAckKeepaliveIntervalMillis);
+                loop.start();
+
+                while (!stopRequested) {
+                    long acked = engine.ackedFsn();
+                    this.ackedFsn = acked;
+                    if (acked >= target) {
+                        outcome = DrainOutcome.SUCCESS;
+                        LOG.info("drainer fully drained slot {} (target={}, acked={})",
+                                slotPath, target, acked);
+                        return;
+                    }
+                    try {
+                        loop.checkError();
+                    } catch (Throwable t) {
+                        if (loop.capabilityGapTerminal() != null) {
+                            // Capability gap mid-drain: recycle the wire, NOT
+                            // the slot. connectWithDurableAckRetry() owns the
+                            // episode budget (16 consecutive gap sweeps /
+                            // wall clock) and drops the sentinel itself if the
+                            // gap persists. The loop's own failed sweep is not
+                            // counted toward the fresh episode -- an off-by-one
+                            // that is immaterial at budget 16.
+                            LOG.warn("drainer slot {}: durable-ack capability gap "
+                                            + "mid-drain ({}), re-entering settle budget",
+                                    slotPath, t.getMessage());
+                            try {
+                                loop.close();
+                            } catch (Throwable ignored) {
+                            }
+                            try {
+                                client.close();
+                            } catch (Throwable ignored) {
+                            }
+                            loop = null;
+                            client = connectWithDurableAckRetry();
+                            if (client == null) {
+                                // outcome already set (FAILED after budget
+                                // exhaustion, or STOPPED); sentinel handled.
+                                return;
+                            }
+                            continue drain;
+                        }
+                        String msg = t.getMessage();
+                        LOG.error("drainer wire error for slot {}: {}", slotPath, msg);
+                        lastErrorMessage = msg;
+                        OrphanScanner.markFailed(slotPath, "wire: " + msg);
+                        outcome = DrainOutcome.FAILED;
+                        return;
+                    }
+                    java.util.concurrent.locks.LockSupport.parkNanos(POLL_NANOS);
                 }
-                java.util.concurrent.locks.LockSupport.parkNanos(POLL_NANOS);
+                // Inner loop exits only on stopRequested; fall through to the
+                // outer condition, which is false for the same reason.
             }
             outcome = DrainOutcome.STOPPED;
         } catch (Throwable t) {

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

@@ -205,6 +205,17 @@ public final class CursorWebSocketSendLoop implements QuietCloseable {
     // up" (looks transient).
     private volatile boolean hasEverConnected;
     private volatile Thread ioThread;
+    // Typed marker for a durable-ack CAPABILITY-GAP terminal: set (before the
+    // terminalError latch, so a checkError() caller that observes the latch is
+    // guaranteed to observe this marker too) when a reconnect sweep threw
+    // QwpDurableAckMismatchException. The orphan drainer consults it to route
+    // a mid-drain capability gap into its budgeted settle-retry
+    // (BackgroundDrainer.connectWithDurableAckRetry) instead of quarantining
+    // the slot on the first sweep; the foreground sender ignores it and keeps
+    // its spec'd loud-fail (sf-client.md section 8.1). Write-once alongside
+    // terminalError: the only writer runs on the I/O thread under the same
+    // first-writer-wins latch.
+    private volatile QwpDurableAckMismatchException capabilityGapTerminal;
     // The latched terminal failure — THE exception every checkError() call
     // rethrows. Write-once for the loop's lifetime: the only writer is
     // recordFatal on the I/O thread (first-writer-wins). The whole
@@ -493,6 +504,23 @@ public void checkError() {
         }
     }
 
+    /**
+     * The typed durable-ack capability-gap terminal, or {@code null} if the
+     * loop's terminal (if any) is a different failure class. Non-null only
+     * after {@link #checkError()} started throwing: the marker is written
+     * before the {@code terminalError} latch, both on the I/O thread.
+     * <p>
+     * Consumer contract: the orphan drainer ({@code BackgroundDrainer})
+     * checks this after a {@code checkError()} throw to decide between
+     * re-entering its budgeted settle-retry (capability gap: the rolling
+     * upgrade may still settle) and quarantining the slot (every other
+     * terminal). Package-private on purpose -- the foreground sender must
+     * not branch on it (spec'd loud-fail, sf-client.md section 8.1).
+     */
+    QwpDurableAckMismatchException capabilityGapTerminal() {
+        return capabilityGapTerminal;
+    }
+
     /**
      * Safety-net variant of {@link #checkError()} for
      * {@code QwpWebSocketSender.close()}: rethrows the latched terminal error
@@ -895,6 +923,13 @@ private void connectLoop(Throwable initial, String phase) {
                 // not SECURITY_ERROR -- this is not an auth failure.
                 LOG.error("durable-ack mismatch during {} -- won't retry: {}",
                         phase, e.getMessage());
+                if (terminalError == null) {
+                    // Mirror recordFatal's first-writer-wins latch: only the
+                    // sweep that owns the terminal may mark the gap, and the
+                    // marker must be visible before the terminalError volatile
+                    // write that checkError() keys on.
+                    capabilityGapTerminal = e;
+                }
                 long fromFsn = engine.ackedFsn() + 1L;
                 long toFsn = Math.max(fromFsn, engine.publishedFsn());
                 SenderError err = new SenderError(