fix(qwp): orphan drainer retries a down server instead of quarantining (Invariant B)

connectWithDurableAckRetry() routed every non-role, non-durable-ack Throwable
-- including "all endpoints unreachable" (server down / network partition) --
to an IMMEDIATE markFailed / .failed sentinel on the first sweep. That
quarantined an orphan slot's un-drained SF data on a transient outage, needing
manual recovery, and was asymmetric with the live sender's background loop
(CursorWebSocketSendLoop.connectLoop, which retries transport errors forever).
It was worst exactly under lazy_connect=true (server down at startup is the
whole point), where it quarantined all orphan data on boot.

Make the drainer classify failures exactly like connectLoop:
- NEW catch (QwpAuthFailedException | WebSocketUpgradeException) -> terminal,
  quarantine immediately (genuinely non-retriable across the cluster).
- catch (Throwable) -> TRANSIENT: back off (capped exponential) and retry
  indefinitely until a primary is reachable, stopRequested, or SF exhaustion;
  never quarantine. WARN throttled to 1/5s so a long/permanent outage stays
  observable without flooding.
- Role reject (all-replica) keeps retry-forever; durable-ack capability gap
  keeps its bounded settle budget (the one deliberate rolling-upgrade tolerance).

Tests:
- testTransportErrorNeverQuarantinesInvariantB: now green (down server retried).
- testNonDurableAckExceptionMarksFailedImmediately -> testTerminalUpgrade
  MarksFailedImmediately: repurposed to a genuine upgrade terminal, since a
  transport error is now retryable. 49 drainer/pool tests pass.

Author: bluestreak01

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

@@ -25,6 +25,8 @@
 package io.questdb.client.cutlass.qwp.client.sf.cursor;
 
 import io.questdb.client.cutlass.http.client.WebSocketClient;
+import io.questdb.client.cutlass.http.client.WebSocketUpgradeException;
+import io.questdb.client.cutlass.qwp.client.QwpAuthFailedException;
 import io.questdb.client.cutlass.qwp.client.QwpDurableAckMismatchException;
 import io.questdb.client.cutlass.qwp.client.QwpIngressRoleRejectedException;
 import io.questdb.client.cutlass.qwp.client.QwpRoleMismatchException;
@@ -172,6 +174,7 @@ public WebSocketClient connectWithDurableAckRetry() {
         // window can last minutes and (Invariant B) is retried indefinitely, so
         // per-attempt logging would flood. Mirrors CursorWebSocketSendLoop.
         long lastReplicaWarnNanos = 0L;
+        long lastTransportWarnNanos = 0L;
         while (!stopRequested) {
             // True only for a genuine durable-ack CAPABILITY gap, which is
             // bounded by the settle budget / attempt cap. A transient all-replica
@@ -181,6 +184,17 @@ public WebSocketClient connectWithDurableAckRetry() {
             boolean boundedByBudget = false;
             try {
                 return clientFactory.reconnect();
+            } catch (QwpAuthFailedException | WebSocketUpgradeException e) {
+                // Genuinely non-retriable across the cluster (auth 401/403, or a
+                // non-421 upgrade reject): waiting will not fix it, so quarantine
+                // immediately -- exactly as the live sender's background loop
+                // (CursorWebSocketSendLoop.connectLoop) halts on these errors.
+                String msg = e.getMessage();
+                LOG.error("drainer terminal upgrade/auth error for slot {}: {}", slotPath, msg);
+                lastErrorMessage = msg;
+                OrphanScanner.markFailed(slotPath, "auth/upgrade: " + msg);
+                outcome = DrainOutcome.FAILED;
+                return null;
             } catch (QwpRoleMismatchException | QwpIngressRoleRejectedException e) {
                 // INVARIANT B: every reachable endpoint is a REPLICA right now.
                 // A replica is promotable and a primary will reappear, so this is
@@ -248,12 +262,21 @@ public WebSocketClient connectWithDurableAckRetry() {
                 LOG.warn("drainer slot {} attempt {}: durable-ack unavailable, retrying after backoff",
                         slotPath, mismatchAttempts);
             } catch (Throwable t) {
-                String msg = t.getMessage();
-                LOG.error("drainer initial connect failed for slot {}: {}", slotPath, msg);
-                lastErrorMessage = msg;
-                OrphanScanner.markFailed(slotPath, "initial connect: " + msg);
-                outcome = DrainOutcome.FAILED;
-                return null;
+                // INVARIANT B: a transport failure -- the whole cluster is
+                // unreachable right now (server down, network partition) -- is
+                // TRANSIENT, exactly as the live sender's background loop treats
+                // it. The server will come back; keep retrying (capped backoff)
+                // until it does, stopRequested, or SF exhaustion. NEVER quarantine
+                // the slot on a transport error. Genuine terminals (auth /
+                // non-421 upgrade / durable-ack capability gap) are handled by the
+                // catches above and still fail fast.
+                lastErrorMessage = t.getMessage();
+                long nowWarn = System.nanoTime();
+                if (nowWarn - lastTransportWarnNanos >= 5_000_000_000L) {
+                    LOG.warn("drainer slot {}: cluster unreachable ({}), retrying after backoff",
+                            slotPath, t.getMessage());
+                    lastTransportWarnNanos = nowWarn;
+                }
             }
             // Backoff before the next sweep. Honor stopRequested by parking in
             // small chunks rather than a single long park so close() doesn't

core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/BackgroundDrainerDurableAckRetryTest.java

@@ -26,6 +26,7 @@
 
 import io.questdb.client.DefaultHttpClientConfiguration;
 import io.questdb.client.cutlass.http.client.WebSocketClient;
+import io.questdb.client.cutlass.http.client.WebSocketUpgradeException;
 import io.questdb.client.network.PlainSocketFactory;
 import io.questdb.client.cutlass.line.LineSenderException;
 import io.questdb.client.cutlass.qwp.client.QwpDurableAckMismatchException;
@@ -40,7 +41,6 @@
 import org.junit.Before;
 import org.junit.Test;
 
-import java.io.IOException;
 import java.nio.file.Paths;
 import java.util.ArrayList;
 import java.util.List;
@@ -211,10 +211,15 @@ public void testNoListenerNoNullPointerOnEscalation() {
     }
 
     @Test
-    public void testNonDurableAckExceptionMarksFailedImmediately() {
+    public void testTerminalUpgradeMarksFailedImmediately() {
         CountingListener listener = new CountingListener();
+        // A genuinely non-retriable upgrade error (non-421 5xx upgrade reject) is
+        // terminal -- waiting will not fix it -- so the drainer quarantines on the
+        // first attempt, exactly like the live sender's background loop halts on
+        // auth/upgrade. A TRANSPORT error, by contrast, is transient and is
+        // retried (see testTransportErrorNeverQuarantinesInvariantB).
         ScriptedFactory factory = ScriptedFactory.alwaysFailing(
-                () -> new IOException("transport down"));
+                () -> new WebSocketUpgradeException(500, null, "server error during upgrade"));
         BackgroundDrainer drainer = newDrainer(factory);
         drainer.setListener(listener);
         WebSocketClient out = drainer.connectWithDurableAckRetry();
@@ -223,10 +228,10 @@ public void testNonDurableAckExceptionMarksFailedImmediately() {
         // Listener must not have been touched — this path doesn't fire either callback.
         assertEquals(0, listener.unavailableAttempts.size());
         assertEquals(0, listener.persistentFailures.get());
-        // Sentinel reason should reflect the non-DA path (initial connect: ...).
+        // Sentinel dropped for a genuine terminal.
         String sentinel = slotPath + "/" + OrphanScanner.FAILED_SENTINEL_NAME;
         assertTrue(Files.exists(sentinel));
-        // The factory must have been invoked exactly once — no retry on this path.
+        // The factory must have been invoked exactly once — no retry on a terminal.
         assertEquals(1, factory.attempts());
     }