fix(qwp): Error escaping the post-upgrade success tail leaks the connected client

The connect walk's catch (Error) arm (1582e3a, cc5e9cc) guarded only
connect/upgrade and the failure arms. After a successful upgrade, the
durable-ack check, success classification, dispatchConnectionEvent and
applyServerBatchSizeLimit ran unguarded before return: an Error thrown
there (the success dispatch allocates the event plus a deque node, and
on a clean first connect lazy-starts the dispatcher thread) escaped
with the fully connected client's fd and native buffers unreachable by
GC.

Wrap the durable-ack-check-to-return tail in the same quiet-close /
rethrow contract as the connect/upgrade arm, extracted into
closeQuietlyOnError so the two arms cannot drift. close() is CAS-gated,
so re-closing after the durable-ack arm's own close is a no-op.

Tests bracket the guarded window at both ends -- first expression
(durable-ack check) and last call (getServerMaxBatchSize) -- so
narrowing the try block later trips one of the two.

Author: bluestreak01

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

@@ -2631,6 +2631,20 @@ public static int effectiveConnectTimeoutMs(boolean background, int configuredMs
      * Production path delegates to {@link WebSocketClientFactory}; tests may
      * install {@link #clientFactoryOverride} to substitute a stub.
      */
+    /**
+     * Best-effort close for a client being abandoned because a JVM Error is
+     * about to be rethrown: under OOM {@code close()} itself can throw, and a
+     * secondary failure must not mask the original Error. {@code close()} is
+     * CAS-gated, so re-closing an already-closed client is a no-op.
+     */
+    private static void closeQuietlyOnError(WebSocketClient client) {
+        try {
+            client.close();
+        } catch (Throwable ignored) {
+            // best-effort; the original Error is what must surface
+        }
+    }
+
     private WebSocketClient newWebSocketClient() {
         java.util.function.Supplier<WebSocketClient> override = clientFactoryOverride;
         if (override != null) {
@@ -2851,77 +2865,90 @@ private WebSocketClient connectWalk(ReconnectSupplier ctx) {
                 // walk. Rethrow: every retry loop upstream (connectWithRetry,
                 // the cursor reconnect loop, BackgroundDrainer) rethrows Error
                 // rather than retrying, so this stays a loud one-shot failure.
-                try {
-                    newClient.close();
-                } catch (Throwable ignored) {
-                    // best-effort; the original Error is what must surface
-                }
+                closeQuietlyOnError(newClient);
                 throw e;
             }
-            if (requestDurableAck && !newClient.isServerDurableAckEnabled()) {
-                newClient.close();
-                hostTracker.recordRoleReject(idx, false, !background);
-                QwpDurableAckMismatchException ackErr = new QwpDurableAckMismatchException(
-                        ep.host, ep.port, null);
-                if (terminalUpgradeError == null) {
-                    terminalUpgradeError = ackErr;
+            // Guard the post-upgrade tail: from here until newClient is
+            // returned, an escaping JVM Error would leak the CONNECTED
+            // client's fd and native buffers -- the same class the
+            // connect/upgrade catch (Error) arm above closes over. The
+            // success-event dispatch is the realistic trigger: it allocates
+            // the SenderConnectionEvent plus a deque node, and on a clean
+            // first connect it is also the dispatcher's first offer(), which
+            // lazy-starts the dispatcher thread (Thread.start() can itself
+            // fail with OOM). Same contract as the arm above: close quietly
+            // (a secondary failure must not mask the original Error) and
+            // rethrow. close() is CAS-gated, so re-closing after the
+            // durable-ack arm's own close is a no-op.
+            try {
+                if (requestDurableAck && !newClient.isServerDurableAckEnabled()) {
+                    newClient.close();
+                    hostTracker.recordRoleReject(idx, false, !background);
+                    QwpDurableAckMismatchException ackErr = new QwpDurableAckMismatchException(
+                            ep.host, ep.port, null);
+                    if (terminalUpgradeError == null) {
+                        terminalUpgradeError = ackErr;
+                    }
+                    lastError = ackErr;
+                    if (!background) {
+                        dispatchConnectionEvent(
+                                SenderConnectionEvent.Kind.ENDPOINT_ATTEMPT_FAILED,
+                                ep.host, ep.port, null, SenderConnectionEvent.NO_PORT,
+                                attemptNumber, roundSeq, ackErr);
+                    }
+                    continue;
                 }
-                lastError = ackErr;
-                if (!background) {
-                    dispatchConnectionEvent(
-                            SenderConnectionEvent.Kind.ENDPOINT_ATTEMPT_FAILED,
-                            ep.host, ep.port, null, SenderConnectionEvent.NO_PORT,
-                            attemptNumber, roundSeq, ackErr);
+                hostTracker.recordSuccess(idx, !background);
+                ctx.previousIdx = idx;
+                if (background) {
+                    // Walk bookkeeping only: recordSuccess feeds the shared health
+                    // tracker and ctx.previousIdx arms this factory's own
+                    // mid-stream-failure handling on its next reconnect. No
+                    // lifecycle event, no CONNECTED/RECONNECTED/FAILED_OVER
+                    // classification state, no producer batch re-sizing -- the
+                    // drainer's lifecycle is observable via
+                    // BackgroundDrainerListener and the drainer counters, never
+                    // the foreground connection-event stream.
+                    return newClient;
                 }
-                continue;
-            }
-            hostTracker.recordSuccess(idx, !background);
-            ctx.previousIdx = idx;
-            if (background) {
-                // Walk bookkeeping only: recordSuccess feeds the shared health
-                // tracker and ctx.previousIdx arms this factory's own
-                // mid-stream-failure handling on its next reconnect. No
-                // lifecycle event, no CONNECTED/RECONNECTED/FAILED_OVER
-                // classification state, no producer batch re-sizing -- the
-                // drainer's lifecycle is observable via
-                // BackgroundDrainerListener and the drainer counters, never
-                // the foreground connection-event stream.
-                return newClient;
-            }
-            int previousLiveIdx = currentEndpointIdx;
-            currentEndpointIdx = idx;
-            // Classify the success. CONNECTED only fires once per sender
-            // lifetime; subsequent successes are RECONNECTED (same endpoint
-            // as before) or FAILED_OVER (different endpoint). hasEverConnected
-            // is set after the classification so the very first success picks
-            // CONNECTED before flipping the flag.
-            SenderConnectionEvent.Kind successKind;
-            String prevHost = null;
-            int prevPort = SenderConnectionEvent.NO_PORT;
-            if (!hasEverConnected) {
-                successKind = SenderConnectionEvent.Kind.CONNECTED;
-                hasEverConnected = true;
-            } else if (previousLiveIdx == idx) {
-                successKind = SenderConnectionEvent.Kind.RECONNECTED;
-            } else {
-                successKind = SenderConnectionEvent.Kind.FAILED_OVER;
-                if (previousLiveIdx >= 0) {
-                    Endpoint prevEp = endpoints.get(previousLiveIdx);
-                    prevHost = prevEp.host;
-                    prevPort = prevEp.port;
+                int previousLiveIdx = currentEndpointIdx;
+                currentEndpointIdx = idx;
+                // Classify the success. CONNECTED only fires once per sender
+                // lifetime; subsequent successes are RECONNECTED (same endpoint
+                // as before) or FAILED_OVER (different endpoint). hasEverConnected
+                // is set after the classification so the very first success picks
+                // CONNECTED before flipping the flag.
+                SenderConnectionEvent.Kind successKind;
+                String prevHost = null;
+                int prevPort = SenderConnectionEvent.NO_PORT;
+                if (!hasEverConnected) {
+                    successKind = SenderConnectionEvent.Kind.CONNECTED;
+                    hasEverConnected = true;
+                } else if (previousLiveIdx == idx) {
+                    successKind = SenderConnectionEvent.Kind.RECONNECTED;
+                } else {
+                    successKind = SenderConnectionEvent.Kind.FAILED_OVER;
+                    if (previousLiveIdx >= 0) {
+                        Endpoint prevEp = endpoints.get(previousLiveIdx);
+                        prevHost = prevEp.host;
+                        prevPort = prevEp.port;
+                    }
                 }
+                dispatchConnectionEvent(
+                        successKind, ep.host, ep.port, prevHost, prevPort,
+                        attemptNumber, roundSeq, null);
+                // Refresh the cap-derived state before returning the new client so
+                // the producer thread observes the new endpoint's advertised
+                // X-QWP-Max-Batch-Size from the next sendRow onwards. Skipping this
+                // on a mid-stream failover leaves the sender sized for the prior
+                // endpoint's cap; an oversize row then escapes the producer-side
+                // guard and trips a wire-level ws-close[1009] downstream.
+                applyServerBatchSizeLimit(newClient.getServerMaxBatchSize());
+                return newClient;
+            } catch (Error e) {
+                closeQuietlyOnError(newClient);
+                throw e;
             }
-            dispatchConnectionEvent(
-                    successKind, ep.host, ep.port, prevHost, prevPort,
-                    attemptNumber, roundSeq, null);
-            // Refresh the cap-derived state before returning the new client so
-            // the producer thread observes the new endpoint's advertised
-            // X-QWP-Max-Batch-Size from the next sendRow onwards. Skipping this
-            // on a mid-stream failover leaves the sender sized for the prior
-            // endpoint's cap; an oversize row then escapes the producer-side
-            // guard and trips a wire-level ws-close[1009] downstream.
-            applyServerBatchSizeLimit(newClient.getServerMaxBatchSize());
-            return newClient;
         }
         // Round walked every endpoint without a success. Surface
         // ALL_ENDPOINTS_UNREACHABLE before any of the typed throws so a

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

@@ -52,6 +52,11 @@
  * rethrows without recording endpoint-health penalties: a JVM failure is not
  * endpoint health data.
  * <p>
+ * Also covers the sibling gap (M5): the post-upgrade SUCCESS tail (durable-ack
+ * check, success-event dispatch, cap re-sizing) used to run unguarded, so an
+ * Error there leaked the fully connected client. A second {@code catch (Error)}
+ * guard now closes it under the same quiet-close / rethrow contract.
+ * <p>
  * Uses the same bare-instance pattern as
  * {@code CursorWebSocketSendLoopJvmErrorTest}: {@code Unsafe.allocateInstance}
  * plus reflective wiring of the fields the connect walk dereferences, with the
@@ -114,6 +119,90 @@ public void testCloseFailureDoesNotMaskOriginalError() throws Exception {
         Assert.assertEquals("close must have been attempted", 1, stub.closeCalls);
     }
 
+    @Test
+    public void testErrorInSuccessTailClosesConnectedClient() throws Exception {
+        // M5 regression: connect and upgrade SUCCEED, then a JVM Error strikes
+        // in the post-upgrade success tail (event dispatch / cap re-sizing,
+        // here injected via getServerMaxBatchSize()). The tail guard must
+        // close the fully CONNECTED client and rethrow the original Error.
+        // recordSuccess bookkeeping stands -- the connect itself was real
+        // health data; only the leak is fixed.
+        QwpWebSocketSender sender = newBareSender();
+        QwpHostHealthTracker tracker = wireEndpoints(sender, 2);
+        List<StubClient> built = new ArrayList<>();
+        OutOfMemoryError oom = new OutOfMemoryError("simulated allocation failure");
+        installFactory(sender, () -> {
+            StubClient c = newStubClient();
+            c.serverMaxBatchSizeError = oom;
+            built.add(c);
+            return c;
+        });
+
+        try {
+            invokeBuildAndConnect(sender);
+            Assert.fail("a JVM Error must propagate out of buildAndConnect");
+        } catch (InvocationTargetException ite) {
+            Assert.assertSame("the original Error must surface", oom, ite.getCause());
+        }
+        Assert.assertEquals("Error must stop the walk on the first attempt", 1, built.size());
+        Assert.assertEquals("connected client must be closed exactly once",
+                1, built.get(0).closeCalls);
+        Assert.assertEquals("success bookkeeping preceded the Error and stands",
+                QwpHostHealthTracker.HostState.HEALTHY, tracker.getState(0));
+        Assert.assertEquals("unattempted endpoint must stay untouched",
+                QwpHostHealthTracker.HostState.UNKNOWN, tracker.getState(1));
+    }
+
+    @Test
+    public void testSuccessTailCloseFailureDoesNotMaskOriginalError() throws Exception {
+        // Same contract as the connect/upgrade arm: the tail guard's cleanup
+        // is best-effort -- a close() failure under memory pressure must not
+        // mask the original Error.
+        QwpWebSocketSender sender = newBareSender();
+        wireEndpoints(sender, 1);
+        OutOfMemoryError oom = new OutOfMemoryError("simulated allocation failure");
+        StubClient stub = newStubClient();
+        stub.serverMaxBatchSizeError = oom;
+        stub.throwOnClose = true;
+        installFactory(sender, () -> stub);
+
+        try {
+            invokeBuildAndConnect(sender);
+            Assert.fail("a JVM Error must propagate out of buildAndConnect");
+        } catch (InvocationTargetException ite) {
+            Assert.assertSame("close() failure must not mask the original Error",
+                    oom, ite.getCause());
+        }
+        Assert.assertEquals("close must have been attempted", 1, stub.closeCalls);
+    }
+
+    @Test
+    public void testErrorAtSuccessTailEntryClosesConnectedClient() throws Exception {
+        // Brackets the tail guard from the front: the durable-ack check is
+        // the FIRST expression after the connect/upgrade try, while the
+        // getServerMaxBatchSize() injection above hits the LAST call before
+        // return. Together they pin the guard to the whole post-upgrade
+        // tail -- narrowing the try block later trips one of the two.
+        QwpWebSocketSender sender = newBareSender();
+        QwpHostHealthTracker tracker = wireEndpoints(sender, 1);
+        setField(sender, "requestDurableAck", true);
+        OutOfMemoryError oom = new OutOfMemoryError("simulated allocation failure");
+        StubClient stub = newStubClient();
+        stub.durableAckCheckError = oom;
+        installFactory(sender, () -> stub);
+
+        try {
+            invokeBuildAndConnect(sender);
+            Assert.fail("a JVM Error must propagate out of buildAndConnect");
+        } catch (InvocationTargetException ite) {
+            Assert.assertSame("the original Error must surface", oom, ite.getCause());
+        }
+        Assert.assertEquals("connected client must be closed exactly once",
+                1, stub.closeCalls);
+        Assert.assertEquals("the Error preceded recordSuccess; no health data recorded",
+                QwpHostHealthTracker.HostState.UNKNOWN, tracker.getState(0));
+    }
+
     @Test
     public void testExceptionPathStillClosesAndWalksAllEndpoints() throws Exception {
         // Seam sanity + behavioral contrast: a plain RuntimeException stays on
@@ -221,6 +310,8 @@ private static final class StubClient extends WebSocketClient {
         int closeCalls;
         Error connectError;
         RuntimeException connectRuntimeError;
+        Error durableAckCheckError;
+        Error serverMaxBatchSizeError;
         boolean throwOnClose;
 
         private StubClient() {
@@ -246,6 +337,26 @@ public void connect(CharSequence host, int port) {
             }
         }
 
+        @Override
+        public boolean isServerDurableAckEnabled() {
+            // First expression of the guarded success tail (evaluated only
+            // when requestDurableAck is wired true).
+            if (durableAckCheckError != null) {
+                throw durableAckCheckError;
+            }
+            return true;
+        }
+
+        @Override
+        public int getServerMaxBatchSize() {
+            // Success-tail injection point: called by applyServerBatchSizeLimit
+            // after a successful connect+upgrade, inside the tail Error guard.
+            if (serverMaxBatchSizeError != null) {
+                throw serverMaxBatchSizeError;
+            }
+            return 0;
+        }
+
         @Override
         public void setConnectTimeout(int connectTimeoutMillis) {
         }