fix(ws): in-callback close() no longer corrupts recv state on the unwinding parse path

CursorWebSocketSendLoop's NACK-recycle (handleServerRejection /
handlePreSendRejection / onClose -> failPaced()/fail() -> connectLoop ->
swapClient -> oldClient.close()) closes the client synchronously while
the I/O thread is still inside that client's tryParseFrame. Control then
unwinds into the post-handler tail, which advanced recvReadPos and
compacted the recv buffer on the just-closed client: recvPos went
negative, and only the accident of disconnect() zeroing positions plus
compactRecvBuffer's remaining > 0 guard stood between that and a
Vect.memmove on the freed recvBufPtr.

Make in-callback close a supported contract instead: tryParseFrame's
tail detects closed and returns PARSE_OK without touching recv state
(the frame was fully dispatched), covering every present and future
frame handler rather than just the cursor send loop. Pin the restored
invariant with an assert in compactRecvBuffer.

Red/green: without the guard the new tests observe recvPos=-6/-4 after
an in-callback close from onBinaryMessage/onClose; with it, positions
stay at disconnect()'s reset. NACK-recycle e2e suites unaffected.

Author: bluestreak01

core/src/main/java/io/questdb/client/cutlass/http/client/WebSocketClient.java

@@ -911,6 +911,12 @@ private void checkConnected() {
     private void compactRecvBuffer() {
         if (recvReadPos > 0) {
             int remaining = recvPos - recvReadPos;
+            // recvPos >= recvReadPos always holds here: a handler-initiated
+            // close() (which zeroes recvPos under our feet) is caught in
+            // tryParseFrame's tail before this method is reached. If this
+            // assert fires, someone reintroduced a post-callback touch of
+            // recv state on a closed/disconnected client.
+            assert remaining >= 0 : "recv buffer positions out of order [recvPos=" + recvPos + ", recvReadPos=" + recvReadPos + ']';
             if (remaining > 0) {
                 Vect.memmove(recvBufPtr, recvBufPtr + recvReadPos, remaining);
             }
@@ -1258,6 +1264,20 @@ private int tryParseFrame(WebSocketFrameHandler handler) {
                     break;
             }
 
+            // A handler callback above may have close()d this client:
+            // CursorWebSocketSendLoop's NACK/close recycle swaps in a new
+            // client and synchronously closes this one (swapClient), then
+            // control unwinds back here. close() -> disconnect() has already
+            // zeroed recvPos/recvReadPos and freed recvBufPtr -- advancing
+            // the read position or compacting would corrupt that state
+            // (negative recvPos today; a memmove on freed memory if the
+            // zeroing ever moved). The frame was fully dispatched, so
+            // in-callback close is a supported contract: report success
+            // and touch nothing. Same-thread, so the closed read is exact.
+            if (closed.get()) {
+                return PARSE_OK;
+            }
+
             // Advance read position
             recvReadPos += consumed;
 

core/src/test/java/io/questdb/client/test/cutlass/http/client/WebSocketClientTest.java

@@ -32,6 +32,7 @@
 import io.questdb.client.network.PlainSocketFactory;
 import io.questdb.client.network.Socket;
 import io.questdb.client.network.SocketReadinessWaiter;
+import io.questdb.client.std.Unsafe;
 import org.junit.Assert;
 import org.junit.Test;
 
@@ -127,6 +128,78 @@ public void testExtractMaxBatchSizeParsesPositive() throws Exception {
         Assert.assertEquals(16 * 1024 * 1024, invokeExtractMaxBatchSize(response));
     }
 
+    /**
+     * A frame handler may close() the client from inside its callback:
+     * CursorWebSocketSendLoop's NACK-recycle path (handleServerRejection /
+     * handlePreSendRejection -> failPaced()/fail() -> connectLoop ->
+     * swapClient -> oldClient.close()) runs synchronously on the I/O thread
+     * while that thread is still inside this client's tryParseFrame. The
+     * post-callback tail must detect the close and touch no recv state:
+     * before the guard, it left recvPos negative on the closed client and
+     * was one close()-reorder away from a memmove on freed memory.
+     */
+    @Test
+    public void testInCallbackCloseFromBinaryHandlerLeavesStateCoherent() throws Exception {
+        assertMemoryLeak(() -> {
+            // Unmasked server->client binary frame: FIN|BINARY, len=4, "abcd"
+            byte[] frame = {(byte) 0x82, 0x04, 'a', 'b', 'c', 'd'};
+            try (FrameFeedWebSocketClient client = new FrameFeedWebSocketClient(frame)) {
+                setUpgradedTrue(client);
+
+                boolean received = client.tryReceiveFrame(new WebSocketFrameHandler() {
+                    @Override
+                    public void onBinaryMessage(long payloadPtr, int payloadLen) {
+                        Assert.assertEquals(4, payloadLen);
+                        client.close();
+                    }
+
+                    @Override
+                    public void onClose(int code, String reason) {
+                    }
+                });
+
+                Assert.assertTrue("frame must be reported as received", received);
+                Assert.assertEquals("recvPos must stay at disconnect()'s reset, not go negative",
+                        0, getIntField(client, "recvPos"));
+                Assert.assertEquals(0, getIntField(client, "recvReadPos"));
+            }
+        });
+    }
+
+    /**
+     * Same contract for the CLOSE-frame branch: onClose handlers routinely
+     * recycle the connection (every WS close is reconnect-eligible), which
+     * closes this client before tryParseFrame's tail runs.
+     */
+    @Test
+    public void testInCallbackCloseFromCloseHandlerLeavesStateCoherent() throws Exception {
+        assertMemoryLeak(() -> {
+            // Unmasked server->client close frame: FIN|CLOSE, len=2, code=1000
+            byte[] frame = {(byte) 0x88, 0x02, 0x03, (byte) 0xE8};
+            try (FrameFeedWebSocketClient client = new FrameFeedWebSocketClient(frame)) {
+                setUpgradedTrue(client);
+
+                boolean received = client.tryReceiveFrame(new WebSocketFrameHandler() {
+                    @Override
+                    public void onBinaryMessage(long payloadPtr, int payloadLen) {
+                        Assert.fail("unexpected binary frame");
+                    }
+
+                    @Override
+                    public void onClose(int code, String reason) {
+                        Assert.assertEquals(1000, code);
+                        client.close();
+                    }
+                });
+
+                Assert.assertTrue("frame must be reported as received", received);
+                Assert.assertEquals("recvPos must stay at disconnect()'s reset, not go negative",
+                        0, getIntField(client, "recvPos"));
+                Assert.assertEquals(0, getIntField(client, "recvReadPos"));
+            }
+        });
+    }
+
     @Test
     public void testRecvOrTimeoutPropagatesNonTimeoutError() throws Exception {
         assertMemoryLeak(() -> {
@@ -258,6 +331,24 @@ private static int invokeExtractMaxBatchSize(String response) throws Exception {
         return (int) m.invoke(null, response);
     }
 
+    private static int getIntField(Object obj, String name) {
+        try {
+            Class<?> clazz = obj.getClass();
+            while (clazz != null) {
+                try {
+                    Field field = clazz.getDeclaredField(name);
+                    field.setAccessible(true);
+                    return field.getInt(obj);
+                } catch (NoSuchFieldException e) {
+                    clazz = clazz.getSuperclass();
+                }
+            }
+            throw new NoSuchFieldException(name);
+        } catch (Exception e) {
+            throw new AssertionError(e);
+        }
+    }
+
     private static void setUpgradedTrue(Object obj) throws Exception {
         Class<?> clazz = obj.getClass();
         while (clazz != null) {
@@ -273,6 +364,94 @@ private static void setUpgradedTrue(Object obj) throws Exception {
         throw new NoSuchFieldException("upgraded");
     }
 
+    /**
+     * WebSocketClient over a socket pre-loaded with one server frame;
+     * sends always succeed. Used to drive a real tryReceiveFrame ->
+     * tryParseFrame -> handler dispatch on the test thread.
+     */
+    private static class FrameFeedWebSocketClient extends WebSocketClient {
+
+        FrameFeedWebSocketClient(byte[] frame) {
+            super(DefaultHttpClientConfiguration.INSTANCE, (nf, log) -> new FrameFeedSocket(frame));
+        }
+
+        @Override
+        protected void ioWait(int timeout, int op) {
+            // no-op: recv delivers data on the first call
+        }
+
+        @Override
+        protected void setupIoWait() {
+            // no-op
+        }
+    }
+
+    /**
+     * Socket that serves a fixed byte sequence from recv() and reports
+     * every send() as fully written (so close-frame sends succeed).
+     */
+    private static class FrameFeedSocket implements Socket {
+        private final byte[] data;
+        private int readPos;
+
+        FrameFeedSocket(byte[] data) {
+            this.data = data;
+        }
+
+        @Override
+        public void close() {
+        }
+
+        @Override
+        public int getFd() {
+            return 0;
+        }
+
+        @Override
+        public boolean isClosed() {
+            return false;
+        }
+
+        @Override
+        public void of(int fd) {
+        }
+
+        @Override
+        public int recv(long bufferPtr, int bufferLen) {
+            int n = Math.min(data.length - readPos, bufferLen);
+            for (int i = 0; i < n; i++) {
+                Unsafe.getUnsafe().putByte(bufferPtr + i, data[readPos + i]);
+            }
+            readPos += n;
+            return n;
+        }
+
+        @Override
+        public int send(long bufferPtr, int bufferLen) {
+            return bufferLen;
+        }
+
+        @Override
+        public void startTlsSession(CharSequence peerName, SocketReadinessWaiter waiter) {
+            throw new UnsupportedOperationException();
+        }
+
+        @Override
+        public boolean supportsTls() {
+            return false;
+        }
+
+        @Override
+        public int tlsIO(int readinessFlags) {
+            return 0;
+        }
+
+        @Override
+        public boolean wantsTlsWrite() {
+            return false;
+        }
+    }
+
     /**
      * Minimal Socket that always returns 0 from recv() (no data available),
      * triggering the ioWait path in recvOrTimeout().