fix(qwp): async/mid-stream never surface a connection-error terminal; drop dead RECONNECT_BUDGET_EXHAUSTED

bluestreak01 · bluestreak01 · commit e9a1e7d7d99f · 2026-07-01T22:32:03.000+01:00
Under Invariant B all senders are SF senders, so a connection error is never
terminal: SYNC initial connect (connectWithRetry) still surfaces it (fail loud),
but ASYNC initial connect and mid-stream reconnect (connectLoop) retry forever
and surface only GENUINE terminals (auth / non-421 upgrade / durable-ack
capability gap) or SF exhaustion. The wall-clock budget give-up was removed from
connectLoop earlier; this removes the now-dead RECONNECT_BUDGET_EXHAUSTED
connection-event kind and fixes the 7 tests that asserted the old behavior.

Main:
- Remove SenderConnectionEvent.Kind.RECONNECT_BUDGET_EXHAUSTED (never fired now)
  and its references in DefaultSenderConnectionListener (switch case + javadoc),
  SenderConnectionListener/SenderConnectionEvent javadoc, and CursorWebSocketSendLoop
  comments.

Tests (all ASYNC or mid-stream -- verified none were SYNC):
- Group 1 (asserted a connection/budget terminal -&gt; now wrong) converted to
  Invariant-B "retries forever, no terminal" guards:
  InitialConnectAsyncTest.testAsyncBudgetExhaustionDeliversToErrorInbox
    -&gt; testAsyncNoServerRetriesForeverNoTerminal
  InitialConnectAsyncTest.testConnectionLostBudgetExhaustionTagsDifferently
    -&gt; testConnectionLostRetriesForeverNoTerminal
  ReconnectTest.testReconnectGivesUpAfterCap -&gt; testReconnectNeverGivesUpInvariantB
- Group 2 (used a terminal to exercise close/facade/ownership) re-pointed from a
  dead-port connection error to a genuine 401 auth terminal (still surfaces in
  async): CloseSafetyNetTest (x2, via TestWebSocketServer.setRejectWithStatus),
  QuestDBFacadeCallbacksTest.testFacadeErrorHandlerReceivesAsyncIngestError,
  CloseOwnershipRaceTest (factory throws QwpAuthFailedException).

67 affected tests pass. SYNC surfacing and its tests are untouched.
diff --git a/core/src/main/java/io/questdb/client/SenderConnectionEvent.java b/core/src/main/java/io/questdb/client/SenderConnectionEvent.java
@@ -96,8 +96,8 @@ public long getAttemptNumber() {
     /**
      * The classified cause of the event, or {@code null} for success/info
      * events ({@link Kind#CONNECTED}, {@link Kind#FAILED_OVER},
-     * {@link Kind#RECONNECTED}). For terminal kinds
-     * ({@link Kind#AUTH_FAILED}, {@link Kind#RECONNECT_BUDGET_EXHAUSTED}) this
+     * {@link Kind#RECONNECTED}). For the terminal kind
+     * ({@link Kind#AUTH_FAILED}) this
      * carries the typed exception that caused the sender to halt.
      */
     @Nullable
@@ -234,14 +234,6 @@ public enum Kind {
          * producer-thread API call surfaces a {@code LineSenderException}.
          * {@link #getCause()} carries the {@code QwpAuthFailedException}.
          */
-        AUTH_FAILED,
-
-        /**
-         * Terminal: the configured reconnect time budget was exhausted without
-         * a successful reconnect. The sender will halt; the next producer-thread
-         * API call surfaces a {@code LineSenderException}. {@link #getCause()}
-         * carries the last observed reconnect error.
-         */
-        RECONNECT_BUDGET_EXHAUSTED
+        AUTH_FAILED
     }
 }
diff --git a/core/src/main/java/io/questdb/client/SenderConnectionListener.java b/core/src/main/java/io/questdb/client/SenderConnectionListener.java
@@ -51,8 +51,8 @@
  * {@link SenderConnectionEvent.Kind#RECONNECTED}) are guaranteed to fire on
  * each transition. Failure events ({@code ENDPOINT_ATTEMPT_FAILED},
  * {@code ALL_ENDPOINTS_UNREACHABLE}) may be coalesced under inbox pressure.
- * Terminal events ({@code AUTH_FAILED}, {@code RECONNECT_BUDGET_EXHAUSTED})
- * fire before the producer-thread {@code LineSenderException} is observable on
+ * The terminal event {@code AUTH_FAILED}
+ * fires before the producer-thread {@code LineSenderException} is observable on
  * the next API call -- so a listener can react sooner than the producer learns
  * via exception, but should not assume the listener fires first under heavy
  * notification load.
diff --git a/core/src/main/java/io/questdb/client/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoop.java b/core/src/main/java/io/questdb/client/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoop.java
@@ -177,11 +177,11 @@ public final class CursorWebSocketSendLoop implements QuietCloseable {
     // alike) is offered to the dispatcher for async delivery to the user's
     // handler. Null disables async delivery entirely; the producer-side
     // typed-throw path is unaffected.
-    // Optional: when non-null, RECONNECT_BUDGET_EXHAUSTED is offered to the
-    // dispatcher for async delivery to the user's listener at the moment
-    // connectLoop gives up. Sender-side fire points (CONNECTED, FAILED_OVER,
-    // ENDPOINT_ATTEMPT_FAILED, AUTH_FAILED, ALL_ENDPOINTS_UNREACHABLE) write
-    // directly to the same dispatcher from QwpWebSocketSender.
+    // Optional: when non-null, sender-side connection events (CONNECTED,
+    // FAILED_OVER, ENDPOINT_ATTEMPT_FAILED, AUTH_FAILED, ALL_ENDPOINTS_UNREACHABLE)
+    // are written to this dispatcher from QwpWebSocketSender. connectLoop itself
+    // no longer emits a terminal budget-exhaustion event (Invariant B: it retries
+    // indefinitely and never gives up on a wall-clock budget).
     private volatile SenderConnectionDispatcher connectionDispatcher;
     private volatile SenderErrorDispatcher errorDispatcher;
     // The send cursor has two coordinate systems:
@@ -667,10 +667,10 @@ public boolean isRunning() {
 
     /**
      * Plug an async-delivery sink for {@link SenderConnectionEvent}
-     * notifications. The loop fires {@code RECONNECT_BUDGET_EXHAUSTED}
-     * through this sink when {@code connectLoop} gives up; other connection
-     * events fire from {@code QwpWebSocketSender.buildAndConnect} directly
-     * into the same dispatcher. Same lifecycle contract as
+     * notifications. Connection events fire from
+     * {@code QwpWebSocketSender.buildAndConnect} directly into this dispatcher;
+     * {@code connectLoop} no longer emits a terminal budget-exhaustion event
+     * (Invariant B: it retries indefinitely). Same lifecycle contract as
      * {@link #setErrorDispatcher}.
      */
     public void setConnectionDispatcher(SenderConnectionDispatcher dispatcher) {
diff --git a/core/src/main/java/io/questdb/client/cutlass/qwp/client/sf/cursor/DefaultSenderConnectionListener.java b/core/src/main/java/io/questdb/client/cutlass/qwp/client/sf/cursor/DefaultSenderConnectionListener.java
@@ -36,9 +36,8 @@
  * transition so silence is never the default -- connect-string-only users
  * still see failover and outage signals in their logs.
  *
- * <p>Terminal kinds ({@code AUTH_FAILED}, {@code RECONNECT_BUDGET_EXHAUSTED})
- * and {@code ALL_ENDPOINTS_UNREACHABLE} fire at WARN level; everything else
- * fires at INFO.
+ * <p>Terminal kind {@code AUTH_FAILED} and {@code ALL_ENDPOINTS_UNREACHABLE}
+ * fire at WARN level; everything else fires at INFO.
  */
 public final class DefaultSenderConnectionListener implements SenderConnectionListener {
 
@@ -52,7 +51,6 @@ private DefaultSenderConnectionListener() {
     public void onEvent(@NotNull SenderConnectionEvent e) {
         switch (e.getKind()) {
             case AUTH_FAILED:
-            case RECONNECT_BUDGET_EXHAUSTED:
             case ALL_ENDPOINTS_UNREACHABLE:
                 LOG.warn("connection event {}", e);
                 break;
diff --git a/core/src/test/java/io/questdb/client/test/QuestDBFacadeCallbacksTest.java b/core/src/test/java/io/questdb/client/test/QuestDBFacadeCallbacksTest.java
@@ -30,6 +30,7 @@
 import io.questdb.client.SenderError;
 import io.questdb.client.SenderErrorHandler;
 import io.questdb.client.test.cutlass.qwp.client.TestPorts;
+import io.questdb.client.test.cutlass.qwp.websocket.TestWebSocketServer;
 import org.jetbrains.annotations.NotNull;
 import org.junit.Assert;
 import org.junit.Test;
@@ -52,6 +53,13 @@
  */
 public class QuestDBFacadeCallbacksTest {
 
+    private static final TestWebSocketServer.WebSocketServerHandler NOOP_HANDLER =
+            new TestWebSocketServer.WebSocketServerHandler() {
+                @Override
+                public void onBinaryMessage(TestWebSocketServer.ClientHandler client, byte[] data) {
+                }
+            };
+
     @Test
     public void testFacadeConnectionListenerReceivesEvents() throws Exception {
         int port = TestPorts.findUnusedPort();
@@ -74,16 +82,24 @@ public void onEvent(@NotNull SenderConnectionEvent event) {
 
     @Test
     public void testFacadeErrorHandlerReceivesAsyncIngestError() throws Exception {
-        int port = TestPorts.findUnusedPort();
-        ErrorInbox inbox = new ErrorInbox();
-        try (QuestDB ignored = QuestDB.builder()
-                .fromConfig(config(port))
-                .errorHandler(inbox)
-                .build()) {
-            Assert.assertTrue(
-                    "facade-wired errorHandler must receive the async budget-exhaustion SenderError",
-                    inbox.await(5, TimeUnit.SECONDS));
-            Assert.assertNotNull("a SenderError must be delivered", inbox.get());
+        // A 401 server produces a genuine auth terminal that surfaces even in
+        // async mode; the facade-wired errorHandler must receive it. (Under
+        // Invariant B a mere connection error would retry forever and never
+        // surface -- only a genuine terminal like auth does.)
+        try (TestWebSocketServer server = new TestWebSocketServer(NOOP_HANDLER)) {
+            server.setRejectWithStatus(401, "Unauthorized");
+            server.start();
+            Assert.assertTrue(server.awaitStart(5, TimeUnit.SECONDS));
+            ErrorInbox inbox = new ErrorInbox();
+            try (QuestDB ignored = QuestDB.builder()
+                    .fromConfig(config(server.getPort()))
+                    .errorHandler(inbox)
+                    .build()) {
+                Assert.assertTrue(
+                        "facade-wired errorHandler must receive the async auth-terminal SenderError",
+                        inbox.await(5, TimeUnit.SECONDS));
+                Assert.assertNotNull("a SenderError must be delivered", inbox.get());
+            }
         }
     }
 
diff --git a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/CloseSafetyNetTest.java b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/CloseSafetyNetTest.java
@@ -30,6 +30,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.test.cutlass.qwp.websocket.TestWebSocketServer;
 import org.jetbrains.annotations.NotNull;
 import org.junit.Assert;
 import org.junit.Rule;
@@ -62,47 +63,59 @@ public class CloseSafetyNetTest {
     public final TemporaryFolder sfDir = TemporaryFolder.builder().assureDeletion().build();
 
     @Test(timeout = 30_000)
-    public void testCloseRethrowsUnsurfacedTerminalWithoutCustomHandler() {
-        // No server, no handler, tight reconnect budget: the I/O thread
-        // latches a never-connected budget-exhaustion terminal that nothing
-        // has surfaced to the user. close() must throw it.
-        Sender sender = Sender.fromConfig(cfg());
-        boolean closed = false;
-        try {
-            awaitLatchedTerminal((QwpWebSocketSender) sender);
+    public void testCloseRethrowsUnsurfacedTerminalWithoutCustomHandler() throws Exception {
+        // A 401 server, no handler: the I/O thread latches a genuine auth
+        // terminal (ws-upgrade-failed / SECURITY_ERROR) that nothing has
+        // surfaced to the user. close() must throw it. (Under Invariant B a
+        // mere connection error would retry forever and never latch -- only a
+        // genuine terminal like auth does.)
+        try (TestWebSocketServer server = new TestWebSocketServer(NOOP_HANDLER)) {
+            server.setRejectWithStatus(401, "Unauthorized");
+            server.start();
+            Assert.assertTrue(server.awaitStart(5, TimeUnit.SECONDS));
+            Sender sender = Sender.fromConfig(cfg(server.getPort()));
+            boolean closed = false;
             try {
-                closed = true;
-                sender.close();
-                Assert.fail("close() must rethrow a terminal error that no synchronous "
-                        + "caller and no custom handler has seen");
-            } catch (LineSenderException e) {
-                String msg = e.getMessage() == null ? "" : e.getMessage();
-                Assert.assertTrue("close() must rethrow the latched terminal: " + msg,
-                        msg.contains("never-connected-budget-exhausted"));
-                Assert.assertTrue("the latched instance is the typed server exception",
-                        e instanceof LineSenderServerException);
-            }
-        } finally {
-            if (!closed) {
-                sender.close();
+                awaitLatchedTerminal((QwpWebSocketSender) sender);
+                try {
+                    closed = true;
+                    sender.close();
+                    Assert.fail("close() must rethrow a terminal error that no synchronous "
+                            + "caller and no custom handler has seen");
+                } catch (LineSenderException e) {
+                    String msg = e.getMessage() == null ? "" : e.getMessage();
+                    Assert.assertTrue("close() must rethrow the latched terminal: " + msg,
+                            msg.contains("ws-upgrade-failed") || msg.contains("401"));
+                    Assert.assertTrue("the latched instance is the typed server exception",
+                            e instanceof LineSenderServerException);
+                }
+            } finally {
+                if (!closed) {
+                    sender.close();
+                }
             }
         }
     }
 
     @Test(timeout = 30_000)
     public void testCloseStaysSilentWhenCustomHandlerAlreadyDelivered() throws Exception {
-        // Same terminal, but the user installed a custom error handler and
+        // Same auth terminal, but the user installed a custom error handler and
         // the dispatcher delivered the error to it. close() must NOT
         // double-signal.
-        ErrorInbox inbox = new ErrorInbox();
-        Sender sender = Sender.builder(cfg())
-                .errorHandler(inbox)
-                .build();
-        Assert.assertTrue("terminal must reach the custom handler within 10s",
-                inbox.await(10, TimeUnit.SECONDS));
-        Assert.assertNotNull(inbox.get());
-        // The handler owns the error now; a rethrow here would double-signal.
-        sender.close();
+        try (TestWebSocketServer server = new TestWebSocketServer(NOOP_HANDLER)) {
+            server.setRejectWithStatus(401, "Unauthorized");
+            server.start();
+            Assert.assertTrue(server.awaitStart(5, TimeUnit.SECONDS));
+            ErrorInbox inbox = new ErrorInbox();
+            Sender sender = Sender.builder(cfg(server.getPort()))
+                    .errorHandler(inbox)
+                    .build();
+            Assert.assertTrue("terminal must reach the custom handler within 10s",
+                    inbox.await(10, TimeUnit.SECONDS));
+            Assert.assertNotNull(inbox.get());
+            // The handler owns the error now; a rethrow here would double-signal.
+            sender.close();
+        }
     }
 
     /**
@@ -120,8 +133,8 @@ private static void awaitLatchedTerminal(QwpWebSocketSender sender) {
         }
     }
 
-    private String cfg() {
-        return "ws::addr=localhost:" + TestPorts.findUnusedPort()
+    private String cfg(int port) {
+        return "ws::addr=localhost:" + port
                 + ";sf_dir=" + sfDir.getRoot().getAbsolutePath()
                 + ";initial_connect_retry=async"
                 + ";reconnect_max_duration_millis=400"
@@ -130,6 +143,13 @@ private String cfg() {
                 + ";close_flush_timeout_millis=0;";
     }
 
+    private static final TestWebSocketServer.WebSocketServerHandler NOOP_HANDLER =
+            new TestWebSocketServer.WebSocketServerHandler() {
+                @Override
+                public void onBinaryMessage(TestWebSocketServer.ClientHandler client, byte[] data) {
+                }
+            };
+
     private static class ErrorInbox implements SenderErrorHandler {
         private final CountDownLatch latch = new CountDownLatch(1);
         private final AtomicReference<SenderError> ref = new AtomicReference<>();
diff --git a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/InitialConnectAsyncTest.java b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/InitialConnectAsyncTest.java
diff --git a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/ReconnectTest.java b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/ReconnectTest.java
diff --git a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/CloseOwnershipRaceTest.java b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/CloseOwnershipRaceTest.java
