Surface startup endpoint-policy failures

endpointPolicyFailureIsTerminal() latched a terminal only for an ORPHAN
drainer, so a FOREGROUND sender retried auth, non-421 upgrade and
durable-ack capability failures forever -- including before it had ever
reached the server. An operator whose credentials are wrong then watched
a mute sender buffer into store-and-forward until it filled and
misreported the cause as "out of space". Two pre-existing tests pin that
startup contract and went red:
QuestDBFacadeCallbacksTest#testFacadeErrorHandlerReceivesAsyncIngestError
asserts the errorHandler receives the async auth terminal, and
QuestDBFacadeDrainerListenerTest asserts close() rethrows the foreground's
latched capability-gap terminal.

Key the terminal off hasEverConnected as well. Connectivity problems are
the caller's problem during startup: SYNC/OFF startup reports them by
throwing from build(); ASYNC has no caller left to throw at, so the
latched terminal reaches the user through SenderErrorHandler and the
close() rethrow. The constructor already seeds hasEverConnected =
client != null, so SYNC/OFF arrives past initialization and never latches,
and swapClient makes the flag sticky: once the wire has been up,
store-and-forward owns the buffered data and every endpoint-policy failure
is a transient to ride out (Invariant B), unchanged. A transport failure
(a dead port) never consulted this predicate and still retries forever.

Rewrite the three tests that asserted the opposite so they pin the
restored contract, and cross-link testAsyncNoServerRetriesForeverNoTerminal
as its transport-half twin.

Reverting the predicate fails both pre-existing tests plus the two
rewritten async-initial ones, while the post-start tests stay green.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: glasstiger

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

@@ -1772,8 +1772,31 @@ private void connectLoop(Throwable initial, String phase, long paceFirstAttemptM
                 phase, elapsedMs, attempts, lastMsg);
     }
 
+    /**
+     * Whether an endpoint-policy failure (auth, non-421 upgrade, durable-ack
+     * capability gap) latches a terminal instead of retrying under Invariant B.
+     * <p>
+     * Two callers own a terminal, for different reasons:
+     * <ul>
+     *   <li>An ORPHAN drainer: a sanctioned terminal that hands the slot back to
+     *       its quarantine owner.</li>
+     *   <li>A sender still INITIALIZING ({@code !hasEverConnected}): connectivity
+     *       problems are the caller's problem during startup and must reach it, so
+     *       an operator learns their credentials are wrong instead of watching a
+     *       silent sender buffer forever. SYNC/OFF startup reports them by throwing
+     *       from {@code build()}; ASYNC startup has no caller left to throw at, so
+     *       the latched terminal reaches the user through {@code SenderErrorHandler}
+     *       and a {@code close()} rethrow. The constructor seeds
+     *       {@code hasEverConnected = client != null}, so SYNC/OFF (which is handed a
+     *       live client) is already past initialization here and never latches.</li>
+     * </ul>
+     * Once a FOREGROUND sender has reached the server even once, initialization is
+     * over and store-and-forward owns the buffered data: every endpoint-policy
+     * failure is then a transient the drainer must ride out, never a producer-fatal
+     * terminal (Invariant B).
+     */
     private boolean endpointPolicyFailureIsTerminal() {
-        return reconnectPolicy == ReconnectPolicy.ORPHAN;
+        return reconnectPolicy == ReconnectPolicy.ORPHAN || !hasEverConnected;
     }
 
     /**
@@ -2287,7 +2310,10 @@ private void swapClient(WebSocketClient newClient) {
         this.client = newClient;
         // Sticky: once the wire is up, we've reached the server at least once
         // for this sender's lifetime. Exposed to the owning sender for
-        // connection-state observability; reconnect policy does not depend on it.
+        // connection-state observability, and it ends initialization: from here
+        // on endpointPolicyFailureIsTerminal() stops latching endpoint-policy
+        // failures on a foreground sender and rides them out instead, because
+        // store-and-forward now owns the buffered data (Invariant B).
         this.hasEverConnected = true;
         if (old != null) {
             try {

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

@@ -58,10 +58,21 @@
 public class InitialConnectAsyncTest {
 
     @Test
-    public void testAsyncAuthFailureRetriesForeverNoTerminal() throws Exception {
-        // Server returns HTTP 401 on every upgrade attempt. Once ASYNC build()
-        // has returned, store-and-forward owns the buffered rows: the I/O thread
-        // must keep retrying rather than terminalizing the producer.
+    public void testAsyncAuthFailureSurfacesTerminal() throws Exception {
+        // Server returns HTTP 401 on every upgrade attempt. A rejection by ENDPOINT
+        // POLICY before the sender has ever reached the server is a startup problem,
+        // not a transient, so it must reach the caller: an operator with the wrong
+        // credentials has to learn that, rather than watch a mute sender buffer into
+        // SF until it fills and misreports the cause as "out of space". SYNC/OFF
+        // startup reports it by throwing from build(); ASYNC has no caller left to
+        // throw at, so it arrives on the SenderErrorHandler instead.
+        //
+        // Contrast testAsyncNoServerRetriesForeverNoTerminal: a dead port is a
+        // TRANSPORT failure -- genuinely transient -- and retries forever even during
+        // startup. And once the wire has been up even once, initialization is over
+        // and store-and-forward owns the data, so the same 401 becomes a transient to
+        // ride out (CursorWebSocketSendLoopForegroundReconnectPolicyTest
+        // #testPostStartAuthFailureRetriesUntilCredentialsRecover).
         try (Always401Fixture fixture = new Always401Fixture()) {
             fixture.start();
             int port = fixture.getPort();
@@ -79,16 +90,17 @@ public void testAsyncAuthFailureRetriesForeverNoTerminal() throws Exception {
                 QwpWebSocketSender wss = (QwpWebSocketSender) sender;
                 awaitAtLeastOneConnectAttempt(wss);
 
-                sender.table("foo").longColumn("v", 1L).atNow();
-                sender.flush();
-
-                Assert.assertFalse(
-                        "async 401 rejection must stay inside the retry loop",
-                        inbox.await(1_500, TimeUnit.MILLISECONDS));
-                Assert.assertNull("no terminal SenderError may be delivered for an async 401",
-                        inbox.get());
+                Assert.assertTrue(
+                        "an async 401 must surface a terminal to the errorHandler",
+                        inbox.await(5, TimeUnit.SECONDS));
+                SenderError err = inbox.get();
+                Assert.assertNotNull("a SenderError must be delivered for an async 401", err);
+                Assert.assertEquals(SenderError.Policy.TERMINAL, err.getAppliedPolicy());
+                Assert.assertEquals(SenderError.Category.SECURITY_ERROR, err.getCategory());
+                Assert.assertTrue("the terminal must name the upgrade rejection, got: "
+                                + err.getServerMessage(),
+                        err.getServerMessage().contains("ws-upgrade-failed"));
                 Assert.assertFalse("no upgrade has succeeded yet", wss.wasEverConnected());
-                awaitReconnectAttemptsAdvance(wss);
             } finally {
                 closeQuietly(sender);
             }
@@ -102,8 +114,9 @@ public void testAsyncNoServerRetriesForeverNoTerminal() throws Exception {
         // (it may appear; the data is safe in SF), so the I/O thread retries
         // forever. reconnect_max_duration_millis is IGNORED as a give-up deadline:
         // no SenderError lands, the sender stays usable, and wasEverConnected()
-        // stays false. Endpoint-policy failures are covered by
-        // testAsyncAuthFailureRetriesForeverNoTerminal.
+        // stays false. This is the TRANSPORT half of the startup contract; the
+        // endpoint-POLICY half, which does surface, is
+        // testAsyncAuthFailureSurfacesTerminal.
         int port = TestPorts.findUnusedPort();
         ErrorInbox inbox = new ErrorInbox();
         String cfg = "ws::addr=localhost:" + port

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

@@ -55,15 +55,17 @@ public class CursorWebSocketSendLoopForegroundReconnectPolicyTest {
     public final TemporaryFolder sfDir = TemporaryFolder.builder().assureDeletion().build();
 
     @Test
-    public void testAsyncInitialAuthFailureRetriesUntilCredentialsRecover() throws Exception {
-        assertAsyncInitialForegroundRecovers(false,
-                () -> new QwpAuthFailedException(401, "localhost", 1));
+    public void testAsyncInitialAuthFailureSurfacesTerminalToTheCaller() throws Exception {
+        assertAsyncInitialForegroundSurfacesTerminal(false,
+                () -> new QwpAuthFailedException(401, "localhost", 1),
+                "ws-upgrade-failed");
     }
 
     @Test
-    public void testAsyncInitialDurableAckMismatchRetriesUntilCapabilityRecovers() throws Exception {
-        assertAsyncInitialForegroundRecovers(true,
-                () -> new QwpDurableAckMismatchException("localhost", 1, "primary"));
+    public void testAsyncInitialDurableAckMismatchSurfacesTerminalToTheCaller() throws Exception {
+        assertAsyncInitialForegroundSurfacesTerminal(true,
+                () -> new QwpDurableAckMismatchException("localhost", 1, "primary"),
+                "durable-ack-mismatch");
     }
 
     @Test
@@ -78,9 +80,18 @@ public void testPostStartDurableAckMismatchRetriesUntilCapabilityRecovers() thro
                 () -> new QwpDurableAckMismatchException("localhost", 1, "primary"));
     }
 
-    private void assertAsyncInitialForegroundRecovers(
+    /**
+     * An endpoint-policy failure before the sender has EVER reached the server is a
+     * startup problem, so it must reach the caller rather than retry silently: an
+     * operator with wrong credentials has to learn that, not watch a mute sender
+     * buffer forever. ASYNC startup has no caller left to throw at, so the terminal
+     * is latched for {@code SenderErrorHandler} delivery and the {@code close()}
+     * rethrow. Post-start the opposite holds -- see {@link #assertForegroundRecovers}.
+     */
+    private void assertAsyncInitialForegroundSurfacesTerminal(
             boolean durableAck,
-            FailureSupplier failureSupplier
+            FailureSupplier failureSupplier,
+            String expectedDetail
     ) throws Exception {
         TestUtils.assertMemoryLeak(() -> {
             DropFirstConnectionHandler handler = new DropFirstConnectionHandler(durableAck, false);
@@ -111,20 +122,18 @@ private void assertAsyncInitialForegroundRecovers(
                     appendFrame(engine, (byte) 1);
                     loop.start();
 
-                    await(() -> factory.attempts() >= 2 || loop.getTerminalError() != null,
-                            "async foreground did not retry the endpoint-policy failure");
-                    Assert.assertNull(
-                            "async initial endpoint-policy failures must not stop the producer",
-                            loop.getTerminalError());
+                    await(() -> loop.getTerminalError() != null,
+                            "an async initial endpoint-policy failure must latch a terminal");
+                    Assert.assertTrue("the terminal must name the endpoint-policy failure, got: "
+                                    + loop.getTerminalError().getMessage(),
+                            loop.getTerminalError().getMessage().contains(expectedDetail));
+                    // Never reached the server, so the terminal is a startup verdict.
                     Assert.assertFalse(loop.hasEverConnected());
-
-                    long target = appendFrame(engine, (byte) 2);
-                    factory.allowConnect();
-                    await(() -> engine.ackedFsn() >= target || loop.getTerminalError() != null,
-                            "async foreground did not deliver after endpoint policy recovered");
-                    Assert.assertNull(loop.getTerminalError());
-                    Assert.assertTrue(loop.hasEverConnected());
-                    Assert.assertEquals(target, engine.ackedFsn());
+                    // Latched on the FIRST rejection: retrying an endpoint-policy
+                    // failure at startup is exactly the silent-buffering regression
+                    // this pins. Attempts cannot climb once running flips false.
+                    Assert.assertEquals("a startup endpoint-policy failure must not be retried",
+                            1, factory.attempts());
                 } finally {
                     factory.allowConnect();
                     loop.close();
@@ -133,6 +142,13 @@ private void assertAsyncInitialForegroundRecovers(
         });
     }
 
+    /**
+     * The post-start twin of {@link #assertAsyncInitialForegroundSurfacesTerminal}:
+     * handing the loop a live client seeds {@code hasEverConnected}, so the sender is
+     * past initialization and store-and-forward owns the buffered data. Every
+     * endpoint-policy failure is then a transient to ride out, never a terminal
+     * (Invariant B).
+     */
     private void assertForegroundRecovers(boolean durableAck, FailureSupplier failureSupplier) throws Exception {
         TestUtils.assertMemoryLeak(() -> {
             DropFirstConnectionHandler handler = new DropFirstConnectionHandler(durableAck, true);