fix(ws): document in-callback close contract; cover fragmented-dispatch close

Review hardening on top of the in-callback close guard:

- WebSocketFrameHandler javadoc now states the contract where implementors
  read it, including the sharp edge: close() frees the buffers the payload
  pointers point into, so payload must be consumed before closing.
- New test pins the CONTINUATION branch: resetFragmentState() runs after
  the handler returns and before the closed-client guard, and must only
  write fields (close() has already freed fragmentBufPtr).
- Test-file standards: inner classes and reflection helpers in alphabetical
  order; getIntField declares throws Exception like its siblings.

Reviewed and accepted without change: in-callback disconnect() (zeroes
positions without setting closed) has no callers anywhere in-tree and is
caught loudly by the compactRecvBuffer assert under -ea; the guard's
closed.get() is one volatile load per parsed frame, noise next to the
recv syscall and frame parse it sits behind.

Author: bluestreak01

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

@@ -32,6 +32,15 @@
  * <p>
  * Thread safety: Callbacks are invoked from the thread that called receiveFrame().
  * Implementations must handle their own synchronization if accessed from multiple threads.
+ * <p>
+ * In-callback close: a callback may {@code close()} the client that is
+ * delivering it — connection recycling does exactly this (see
+ * CursorWebSocketSendLoop, whose NACK/close handling swaps in a new client
+ * and synchronously closes the old one before the callback returns). The
+ * client detects the close when the callback returns, touches no further
+ * receive state, and reports the frame as received. Note that
+ * {@code close()} frees the buffers the payload pointers point into: read
+ * everything you need from the payload BEFORE closing.
  */
 public interface WebSocketFrameHandler {
 

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

@@ -200,6 +200,54 @@ public void onClose(int code, String reason) {
         });
     }
 
+    /**
+     * Fragmented-message variant: the CONTINUATION branch runs
+     * resetFragmentState() after the handler returns and before the
+     * closed-client guard. Pin that an in-callback close from the
+     * reassembled-message dispatch stays safe there too: close() frees
+     * fragmentBufPtr, so resetFragmentState() must only write fields.
+     */
+    @Test
+    public void testInCallbackCloseFromFragmentedMessageLeavesStateCoherent() throws Exception {
+        assertMemoryLeak(() -> {
+            // Two unmasked server->client frames: non-FIN BINARY "ab",
+            // then FIN CONTINUATION "cd" -> one reassembled message "abcd"
+            byte[] frames = {
+                    0x02, 0x02, 'a', 'b',
+                    (byte) 0x80, 0x02, 'c', 'd'
+            };
+            try (FrameFeedWebSocketClient client = new FrameFeedWebSocketClient(frames)) {
+                setUpgradedTrue(client);
+                final int[] messages = {0};
+                WebSocketFrameHandler handler = new WebSocketFrameHandler() {
+                    @Override
+                    public void onBinaryMessage(long payloadPtr, int payloadLen) {
+                        messages[0]++;
+                        Assert.assertEquals(4, payloadLen);
+                        client.close();
+                    }
+
+                    @Override
+                    public void onClose(int code, String reason) {
+                        Assert.fail("unexpected close frame");
+                    }
+                };
+
+                // First call consumes the initial fragment: parsed and
+                // buffered, no dispatch yet.
+                Assert.assertTrue(client.tryReceiveFrame(handler));
+                Assert.assertEquals(0, messages[0]);
+
+                // Second call reassembles and dispatches; the handler
+                // closes the client in-callback.
+                Assert.assertTrue(client.tryReceiveFrame(handler));
+                Assert.assertEquals(1, messages[0]);
+                Assert.assertEquals(0, getIntField(client, "recvPos"));
+                Assert.assertEquals(0, getIntField(client, "recvReadPos"));
+            }
+        });
+    }
+
     @Test
     public void testRecvOrTimeoutPropagatesNonTimeoutError() throws Exception {
         assertMemoryLeak(() -> {
@@ -325,30 +373,26 @@ public void testSendPingDoesNotClobberSendBuffer() throws Exception {
         });
     }
 
+    private static int getIntField(Object obj, String name) throws Exception {
+        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);
+    }
+
     private static int invokeExtractMaxBatchSize(String response) throws Exception {
         Method m = WebSocketClient.class.getDeclaredMethod("extractMaxBatchSize", String.class);
         m.setAccessible(true);
         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) {
@@ -365,38 +409,10 @@ private static void setUpgradedTrue(Object obj) throws Exception {
     }
 
     /**
-     * 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).
+     * Minimal Socket that always returns 0 from recv() (no data available),
+     * triggering the ioWait path in recvOrTimeout().
      */
-    private static class FrameFeedSocket implements Socket {
-        private final byte[] data;
-        private int readPos;
-
-        FrameFeedSocket(byte[] data) {
-            this.data = data;
-        }
+    private static class FakeSocket implements Socket {
 
         @Override
         public void close() {
@@ -418,17 +434,12 @@ 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;
+            return 0;
         }
 
         @Override
         public int send(long bufferPtr, int bufferLen) {
-            return bufferLen;
+            return 0;
         }
 
         @Override
@@ -453,10 +464,16 @@ public boolean wantsTlsWrite() {
     }
 
     /**
-     * Minimal Socket that always returns 0 from recv() (no data available),
-     * triggering the ioWait path in recvOrTimeout().
+     * Socket that serves a fixed byte sequence from recv() and reports
+     * every send() as fully written (so close-frame sends succeed).
      */
-    private static class FakeSocket implements Socket {
+    private static class FrameFeedSocket implements Socket {
+        private final byte[] data;
+        private int readPos;
+
+        FrameFeedSocket(byte[] data) {
+            this.data = data;
+        }
 
         @Override
         public void close() {
@@ -478,12 +495,17 @@ public void of(int fd) {
 
         @Override
         public int recv(long bufferPtr, int bufferLen) {
-            return 0;
+            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 0;
+            return bufferLen;
         }
 
         @Override
@@ -507,6 +529,28 @@ public boolean wantsTlsWrite() {
         }
     }
 
+    /**
+     * WebSocketClient over a socket pre-loaded with canned server frames;
+     * sends always succeed. Used to drive a real tryReceiveFrame ->
+     * tryParseFrame -> handler dispatch on the test thread.
+     */
+    private static class FrameFeedWebSocketClient extends WebSocketClient {
+
+        FrameFeedWebSocketClient(byte[] frames) {
+            super(DefaultHttpClientConfiguration.INSTANCE, (nf, log) -> new FrameFeedSocket(frames));
+        }
+
+        @Override
+        protected void ioWait(int timeout, int op) {
+            // no-op: recv delivers data on the first call
+        }
+
+        @Override
+        protected void setupIoWait() {
+            // no-op
+        }
+    }
+
     /**
      * WebSocketClient subclass with a fake socket that always returns 0
      * from recv(), forcing the ioWait path in recvOrTimeout().