From d3de6b02a5a70d5bf04b186f82fb4a5774d7388b Mon Sep 17 00:00:00 2001 From: Jaromir Hamala Date: Wed, 4 Nov 2026 13:37:05 +0100 Subject: [PATCH 01/13] perf(ws): send first, poll only on backpressure doSend() waited on EPOLLOUT before every send(), costing epoll_ctl + epoll_wait per WebSocket frame even when the socket was writable. Try send() first and fall back to ioWait only when the socket is full (send returns ERETRY/0), mirroring recvOrDie's recv-first pattern. --- .../cutlass/http/client/WebSocketClient.java | 11 ++- .../http/client/WebSocketClientTest.java | 70 +++++++++++++++++-- 2 files changed, 73 insertions(+), 8 deletions(-) diff --git a/core/src/main/java/io/questdb/client/cutlass/http/client/WebSocketClient.java b/core/src/main/java/io/questdb/client/cutlass/http/client/WebSocketClient.java index f4d43599..06e49733 100644 --- a/core/src/main/java/io/questdb/client/cutlass/http/client/WebSocketClient.java +++ b/core/src/main/java/io/questdb/client/cutlass/http/client/WebSocketClient.java @@ -1145,18 +1145,23 @@ private void doConnect(CharSequence host, int port) { private void doSend(long ptr, int len, int timeout) { long startTime = System.nanoTime(); recvDrained = false; + // Send first, poll only on backpressure (mirrors recvOrDie): the + // socket is non-blocking, and a pre-send ioWait would cost an + // epoll_ctl + epoll_wait per frame even when it is writable. while (len > 0) { - int remainingTimeout = getRemainingTimeOrThrow(timeout, startTime); - ioWait(remainingTimeout, IOOperation.WRITE); int sent = dieIfNegative(socket.send(ptr, len)); while (socket.wantsTlsWrite()) { - remainingTimeout = getRemainingTimeOrThrow(timeout, startTime); + int remainingTimeout = getRemainingTimeOrThrow(timeout, startTime); ioWait(remainingTimeout, IOOperation.WRITE); dieIfNegative(socket.tlsIO(Socket.WRITE_FLAG)); } if (sent > 0) { ptr += sent; len -= sent; + } else { + // send returned ERETRY (0): socket full, park until writable + int remainingTimeout = getRemainingTimeOrThrow(timeout, startTime); + ioWait(remainingTimeout, IOOperation.WRITE); } } } diff --git a/core/src/test/java/io/questdb/client/test/cutlass/http/client/WebSocketClientTest.java b/core/src/test/java/io/questdb/client/test/cutlass/http/client/WebSocketClientTest.java index d0335886..4cf24c6f 100644 --- a/core/src/test/java/io/questdb/client/test/cutlass/http/client/WebSocketClientTest.java +++ b/core/src/test/java/io/questdb/client/test/cutlass/http/client/WebSocketClientTest.java @@ -30,6 +30,7 @@ import io.questdb.client.cutlass.http.client.WebSocketClient; import io.questdb.client.cutlass.http.client.WebSocketFrameHandler; import io.questdb.client.cutlass.http.client.WebSocketSendBuffer; +import io.questdb.client.network.IOOperation; import io.questdb.client.network.PlainSocketFactory; import io.questdb.client.network.Socket; import io.questdb.client.network.SocketReadinessWaiter; @@ -40,6 +41,7 @@ import java.lang.reflect.Field; import java.lang.reflect.Method; +import java.util.Arrays; import java.util.concurrent.CyclicBarrier; import java.util.concurrent.atomic.AtomicReference; @@ -349,6 +351,21 @@ public void testSendBinaryAssemblesTwoPayloadSlicesIntoOneMaskedFrame() throws E }); } + /** + * doSend() must try send() first and poll for writability only when the + * socket reports full (send returns 0 / ERETRY); a pre-send ioWait costs + * an epoll_ctl + epoll_wait per frame even when the socket is writable. + */ + @Test + public void testSendPollsForWritabilityOnlyOnBackpressure() throws Exception { + assertMemoryLeak(() -> { + // whole frame accepted by the first send(): no poll + assertSendBinaryWriteWaits(new CapturingSocket(), 0); + // 2 bytes accepted, socket full once, then the rest: exactly one poll + assertSendBinaryWriteWaits(new CapturingSocket(2, 0), 1); + }); + } + @Test public void testSendCloseFrameDoesNotClobberSendBuffer() throws Exception { assertMemoryLeak(() -> { @@ -410,6 +427,31 @@ public void testSendPingDoesNotClobberSendBuffer() throws Exception { }); } + private static void assertSendBinaryWriteWaits(CapturingSocket socket, int expectedWriteWaits) throws Exception { + long payload = Unsafe.malloc(7, MemoryTag.NATIVE_DEFAULT); + try (CapturingWebSocketClient client = new CapturingWebSocketClient(socket)) { + setUpgradedTrue(client); + for (int i = 0; i < 7; i++) { + Unsafe.getUnsafe().putByte(payload + i, (byte) (i + 1)); + } + + client.sendBinary(payload, 7); + + byte[] frame = socket.sent; + Assert.assertEquals(2 + 4 + 7, frame.length); + Assert.assertEquals((byte) 0x82, frame[0]); + Assert.assertEquals((byte) (0x80 | 7), frame[1]); + byte[] unmasked = new byte[7]; + for (int i = 0; i < unmasked.length; i++) { + unmasked[i] = (byte) (frame[6 + i] ^ frame[2 + (i & 3)]); + } + Assert.assertArrayEquals(new byte[]{1, 2, 3, 4, 5, 6, 7}, unmasked); + Assert.assertEquals(expectedWriteWaits, client.writeWaits); + } finally { + Unsafe.free(payload, 7, MemoryTag.NATIVE_DEFAULT); + } + } + private static int getIntField(Object obj, String name) throws Exception { Class clazz = obj.getClass(); while (clazz != null) { @@ -500,20 +542,35 @@ public boolean wantsTlsWrite() { } } + /** + * Accumulates every byte send() accepts. The i-th send() call accepts at + * most acceptLimits[i] bytes (0 = socket full); calls past the script + * accept everything. + */ private static class CapturingSocket extends FakeSocket { - private byte[] sent; + private final int[] acceptLimits; + private byte[] sent = new byte[0]; + private int sendCalls; + + CapturingSocket(int... acceptLimits) { + this.acceptLimits = acceptLimits; + } @Override public int send(long bufferPtr, int bufferLen) { - sent = new byte[bufferLen]; - for (int i = 0; i < bufferLen; i++) { - sent[i] = Unsafe.getUnsafe().getByte(bufferPtr + i); + int n = sendCalls < acceptLimits.length ? Math.min(acceptLimits[sendCalls], bufferLen) : bufferLen; + sendCalls++; + int offset = sent.length; + sent = Arrays.copyOf(sent, offset + n); + for (int i = 0; i < n; i++) { + sent[offset + i] = Unsafe.getUnsafe().getByte(bufferPtr + i); } - return bufferLen; + return n; } } private static class CapturingWebSocketClient extends TestWebSocketClient { + private int writeWaits; CapturingWebSocketClient(CapturingSocket socket) { super(DefaultHttpClientConfiguration.INSTANCE, (nf, log) -> socket); @@ -521,6 +578,9 @@ private static class CapturingWebSocketClient extends TestWebSocketClient { @Override protected void ioWait(int timeout, int op) { + if (op == IOOperation.WRITE) { + writeWaits++; + } } @Override From af510873219d1533fb8b5c330f0de2215509807a Mon Sep 17 00:00:00 2001 From: Jaromir Hamala Date: Wed, 4 Nov 2026 14:02:18 +0100 Subject: [PATCH 02/13] perf(qwp): batch published frames into one send() per I/O loop pass trySendOne sent one WebSocket frame per send() syscall even with many frames already published. Walk all fully published frames up to the send buffer's capacity, encode them back to back (exact-size masked headers), and flush with one doSend. ACK fairness is now bounded by the byte cap rather than by one frame per iteration. Per-frame checks and bookkeeping are unchanged; early exits flush appended frames first so behaviour matches the old path where earlier frames were already sent. --- .../cutlass/http/client/WebSocketClient.java | 34 ++ .../http/client/WebSocketSendBuffer.java | 26 ++ .../sf/cursor/CursorWebSocketSendLoop.java | 354 ++++++++++-------- .../http/client/WebSocketClientTest.java | 57 +++ ...kgroundDrainerInterruptedTeardownTest.java | 2 +- ...WebSocketSendLoopBlockedSendCloseTest.java | 7 +- ...WebSocketSendLoopCatchUpAlignmentTest.java | 4 +- ...SendLoopForegroundReconnectPolicyTest.java | 2 +- .../CursorWebSocketSendLoopIdleWaitTest.java | 103 ++++- ...CursorWebSocketSendLoopOrphanTailTest.java | 8 +- ...ursorWebSocketSendLoopPoisonFrameTest.java | 8 +- ...rsorWebSocketSendLoopRotationRaceTest.java | 8 +- ...sorWebSocketSendLoopTornDictGuardTest.java | 58 ++- 13 files changed, 493 insertions(+), 178 deletions(-) diff --git a/core/src/main/java/io/questdb/client/cutlass/http/client/WebSocketClient.java b/core/src/main/java/io/questdb/client/cutlass/http/client/WebSocketClient.java index 06e49733..dd6ca2b1 100644 --- a/core/src/main/java/io/questdb/client/cutlass/http/client/WebSocketClient.java +++ b/core/src/main/java/io/questdb/client/cutlass/http/client/WebSocketClient.java @@ -241,6 +241,40 @@ public WebSocketClient(HttpClientConfiguration configuration, SocketFactory sock } } + /** + * Appends one binary frame to the send batch, back to back with any frame + * already appended. Nothing reaches the wire until {@link #batchFlush()}. + * Grows the send buffer if the frame does not fit; a caller bounding the + * batch checks {@link #batchCapacity()} first. The batch shares the send + * buffer with {@link #sendBinary}, so do not interleave the two. + */ + public void batchAppendBinary(long dataPtr, int length) { + checkConnected(); + sendBuffer.appendMaskedBinaryFrame(dataPtr, length); + } + + /** + * Bytes the send batch holds without growing the send buffer. + */ + public int batchCapacity() { + return sendBuffer.getCapacity(); + } + + /** + * Writes every frame appended since the last flush with one send. No-op + * when nothing was appended. The batch is empty afterwards, even on failure. + */ + public void batchFlush() { + int len = sendBuffer.getWritePos(); + if (len > 0) { + try { + doSend(sendBuffer.getBufferPtr(), len, defaultTimeout); + } finally { + sendBuffer.reset(); + } + } + } + @Override public void close() { // CAS gate: exactly one closer runs the teardown below. Closers can be diff --git a/core/src/main/java/io/questdb/client/cutlass/http/client/WebSocketSendBuffer.java b/core/src/main/java/io/questdb/client/cutlass/http/client/WebSocketSendBuffer.java index 37fe03a6..825c52a5 100644 --- a/core/src/main/java/io/questdb/client/cutlass/http/client/WebSocketSendBuffer.java +++ b/core/src/main/java/io/questdb/client/cutlass/http/client/WebSocketSendBuffer.java @@ -110,6 +110,32 @@ public WebSocketSendBuffer(int initialCapacity, int maxBufferSize) { this.payloadStartOffset = 0; } + /** + * Appends one complete masked binary frame at the write position. Unlike + * {@link #beginFrame()}/{@link #endFrame(int)}, which reserve + * {@code MAX_HEADER_SIZE} and leave a gap before a shorter header, the + * header here is exact-size, so consecutive frames are contiguous and the + * whole buffer can go out in one write. + * + * @param payloadPtr pointer to the payload + * @param payloadLen payload length + */ + public void appendMaskedBinaryFrame(long payloadPtr, int payloadLen) { + int headerSize = WebSocketFrameWriter.headerSize(payloadLen, true); + long requiredCapacity = (long) writePos + headerSize + payloadLen; + if (requiredCapacity > bufCapacity) { + grow(requiredCapacity); + } + int maskKey = rnd.nextInt(); + long payload = bufPtr + writePos + WebSocketFrameWriter.writeHeader( + bufPtr + writePos, true, WebSocketOpcode.BINARY, payloadLen, maskKey); + if (payloadLen > 0) { + Vect.memcpy(payload, payloadPtr, payloadLen); + WebSocketFrameWriter.maskPayload(payload, payloadLen, maskKey); + } + writePos += headerSize + payloadLen; + } + /** * Begins a new WebSocket frame. Reserves space for the maximum header size. * The opcode is specified later when ending the frame via {@link #endFrame(int)}. 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 index 6d429bcb..8b77d65f 100644 --- 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 @@ -42,6 +42,7 @@ import io.questdb.client.cutlass.qwp.client.WebSocketResponse; import io.questdb.client.cutlass.qwp.protocol.QwpConstants; import io.questdb.client.cutlass.qwp.websocket.WebSocketCloseCode; +import io.questdb.client.cutlass.qwp.websocket.WebSocketFrameWriter; import io.questdb.client.std.CharSequenceLongHashMap; import io.questdb.client.std.MemoryTag; import io.questdb.client.std.QuietCloseable; @@ -367,7 +368,7 @@ public final class CursorWebSocketSendLoop implements QuietCloseable { // ALWAYS active -- in memory mode, in disk mode, and (critically) even when the // per-slot persisted dictionary failed to open. sentDictCount is this loop's model // of how many ids the CURRENT server has been told about, and the torn-dict guard - // in trySendOne reads it, so it must track the wire in every mode; gating it on + // in trySendBatch reads it, so it must track the wire in every mode; gating it on // engine.isDeltaDictEnabled() is what once froze it at 0 and terminal'd a slot // whose frames replay perfectly from id 0 (see the constructor). On a recovered / // orphan-drained slot the constructor SEEDS sentDict* from the persisted @@ -434,8 +435,8 @@ public final class CursorWebSocketSendLoop implements QuietCloseable { // the upgrade + catch-up then closes) would be mistaken for a deterministic // head-frame rejection and escalate to a PROTOCOL_VIOLATION terminal -- breaking // the store-and-forward "retry a transient outage forever" contract. Reset per - // connection in setWireBaselineWithCatchUp; set in trySendOne after a successful - // send. + // connection in setWireBaselineWithCatchUp; set in trySendBatch when a data frame + // joins the batch (the flush follows in the same call; a failed flush recycles). private boolean dataFrameSentThisConnection; private volatile WebSocketClient client; // Optional: when non-null, every server-rejection error (retriable and @@ -559,7 +560,7 @@ public final class CursorWebSocketSendLoop implements QuietCloseable { // rethrows always deliver the same instance. private volatile LineSenderException terminalError; // Wall clock of the last outbound activity on the wire -- a sent frame - // (trySendOne) or a keepalive PING (sendPingIfDue). + // (trySendBatch) or a keepalive PING (sendPingIfDue). // Throttles the keepalive PINGs so they fire only when the // configured interval has elapsed since the most recent outbound event. // Set when the I/O loop starts. @@ -575,7 +576,7 @@ public final class CursorWebSocketSendLoop implements QuietCloseable { // value confirms replay is working; a sustained nonzero rate means // the connection is flapping and replay is doing real work each cycle. // Snapshotted when an established connection enters reconnect; cleared - // back to -1 once trySendOne has caught up past it. Used to count replay + // back to -1 once trySendBatch has caught up past it. Used to count replay // frames without a per-frame branch on the steady-state path. private long replayTargetFsn = -1L; // Recovered orphaned deferred tail: frames [orphanSkipStartFsn .. @@ -583,7 +584,7 @@ public final class CursorWebSocketSendLoop implements QuietCloseable { // -- a transaction whose commit was never published (producer crashed or // closed mid-transaction). They must never reach the wire: a // commit-bearing frame from THIS session would commit them server-side - // and resurrect a partial transaction. trySendOne stops the cursor at the + // and resurrect a partial transaction. trySendBatch stops the cursor at the // tail; tryRetireOrphanTail retires it with a cumulative self-acknowledge // once everything below is server-acked. -1 when none/retired. Written by // the constructor and start() (user thread, before the I/O thread exists) @@ -2760,15 +2761,15 @@ private void ioLoop() { // Snapshot before the send attempt: a publish after this point is // caught by the re-check below, so the idle decision cannot race it. long publishedBeforeWork = engine.publishStamp(); - boolean didWork = trySendOne(); + boolean didWork = trySendBatch(); // 1. Try to send next frame(s). // 2. Try to receive ACKs. if (tryReceiveAcks()) { didWork = true; } - // A reconnect anywhere in this iteration (inside trySendOne, + // A reconnect anywhere in this iteration (inside trySendBatch, // tryReceiveAcks or a previous awaitIdle) repositioned the cursor: - // go straight back to trySendOne instead of blocking. + // go straight back to trySendBatch instead of blocking. if (recycledThisIteration) { recycledThisIteration = false; didWork = true; @@ -2794,7 +2795,7 @@ private void ioLoop() { // setIoWakeup), close() rings it, or the next PING is due. // No spinning: an idle connection costs zero CPU. A PING that // failed in 3/4 reconnected and rewound the cursor, which set - // recycledThisIteration again: go back to trySendOne instead. + // recycledThisIteration again: go back to trySendBatch instead. if (!didWork && running && !recycledThisIteration) { ioParked = true; try { @@ -2914,7 +2915,7 @@ private void positionCursorAt(long targetFsn) { // barrier. If it saw the new frame bytes, the frameCount write that // makes targetFsn discoverable is also visible, so a second lookup // must now find it. If the producer publishes later, sendOffset is - // still at the old tip and trySendOne() will send the frame normally. + // still at the old tip and trySendBatch() will send the frame normally. seg = engine.pinSegmentContaining(targetFsn); if (seg != null) { positionCursorInSegment(seg, targetFsn); @@ -3127,7 +3128,7 @@ private void snapshotReplayTarget() { * Reset wire state for a fresh connection: install the new client, * realign {@code fsnAtZero} to the next unacked FSN, restart wire * sequencing from 0, and reposition the cursor so the next - * {@link #trySendOne} call replays the first unacked frame. + * {@link #trySendBatch} call replays the first unacked frame. */ private void swapClient(WebSocketClient newClient) { WebSocketClient old = this.client; @@ -3270,7 +3271,7 @@ private void accumulateSentDict(long payloadAddr, int payloadLen, int deltaStart // The mirror holds ids [0, sentDictCount). Accumulate ONLY the part of this // frame's delta [deltaStart, deltaStart+deltaCount) that extends past the // tip -- ids [sentDictCount, deltaStart+deltaCount). Cases: - // - deltaStart > sentDictCount: a gap. trySendOne's torn-dictionary guard + // - deltaStart > sentDictCount: a gap. trySendBatch's torn-dictionary guard // rejects it before send, so it never reaches here; bail defensively // rather than accumulate past a hole. // - deltaEnd <= sentDictCount: a pure replay/overlap we already hold -- @@ -3834,8 +3835,8 @@ public void setWireBaselineWithCatchUpForTest(long replayStart) { } @TestOnly - public boolean trySendOneForTest() { - return trySendOne(); + public boolean trySendBatchForTest() { + return trySendBatch(); } private void ensureCatchUpFrameCapacity(int required) { @@ -3877,157 +3878,206 @@ private boolean tryReceiveAcks() { /** * Returns true if at least one frame was sent (caller skips the park). - * Bounded: sends at most one frame per call so the ACK side gets - * scheduling fairness. + * Sends every fully published frame of the current segment in one write, + * up to the send buffer's capacity, so the ACK side still gets a turn + * every buffer's worth of bytes and the buffer never grows mid-batch (a + * single frame larger than the buffer goes alone and grows it). + *

+ * Every per-frame check runs per frame, in the old order. A frame that is + * not a plain send (orphan tail, corrupt length, torn dictionary, segment + * hand-off) ends a non-empty batch BEFORE it: the batch is flushed and the + * next call meets that frame with an empty batch, exactly as the + * one-frame-per-call path did after its previous frame went out. */ - private boolean trySendOne() { - if (orphanSkipTipFsn >= 0 && fsnAtZero + nextWireSeq >= orphanSkipStartFsn) { - // The send cursor reached the orphaned deferred tail. Its frames - // belong to an aborted transaction and must never be transmitted - // (a commit-bearing frame from this session would commit them -- - // partial-transaction resurrection). - if (!tryRetireOrphanTail()) { - // Frames below the tail still await server acks; keep ticking - // (tryReceiveAcks runs every loop iteration) until the ack - // watermark reaches the tail's lower edge. - return false; - } - if (nextWireSeq == 0) { - // Reached only when the tail was NOT retirable at connection setup -- - // both setup sites (swapClient, positionCursorForStart) call - // tryRetireOrphanTail first -- which means frames below it still needed - // acks, which means they were SENT on this connection. So nextWireSeq is - // never 0 here in practice and this arm is effectively dead; it is kept - // as a cheap correctness guard rather than removed. - // - // NB the mapping is untouched only while NO wire sequence has been - // consumed. A dictionary catch-up consumes sequences 0..n-1 before any - // data frame, so after one this test is false and the recycle below -- - // which re-anchors the mapping from scratch -- is the correct path, not - // an avoidable cost. - try { - positionCursorForStart(); - } catch (CatchUpSendException e) { - // Re-anchor's catch-up send failed. fail() here is a fresh, - // non-re-entrant connectLoop entry from the I/O loop body -- - // the same recovery a normal trySendOne send failure takes. - // Preserve the wrapper on a cap gap so connectLoop's - // isCatchUpCapGap keeps the orphan settle episode (attempt count - // + dwell anchor) alive across the re-anchor recycle; an ordinary - // failure unwraps to the raw cause and restarts the episode, like - // any normal send failure. - fail(isCatchUpCapGap(e) ? e : e.getCause()); + private boolean trySendBatch() { + int batchFrames = 0; + int batchReplayed = 0; + long batchBytes = 0; + while (true) { + if (orphanSkipTipFsn >= 0 && fsnAtZero + nextWireSeq >= orphanSkipStartFsn) { + if (batchFrames > 0) { + break; + } + // The send cursor reached the orphaned deferred tail. Its frames + // belong to an aborted transaction and must never be transmitted + // (a commit-bearing frame from this session would commit them -- + // partial-transaction resurrection). + if (!tryRetireOrphanTail()) { + // Frames below the tail still await server acks; keep ticking + // (tryReceiveAcks runs every loop iteration) until the ack + // watermark reaches the tail's lower edge. return false; } - return true; + if (nextWireSeq == 0) { + // Reached only when the tail was NOT retirable at connection setup -- + // both setup sites (swapClient, positionCursorForStart) call + // tryRetireOrphanTail first -- which means frames below it still needed + // acks, which means they were SENT on this connection. So nextWireSeq is + // never 0 here in practice and this arm is effectively dead; it is kept + // as a cheap correctness guard rather than removed. + // + // NB the mapping is untouched only while NO wire sequence has been + // consumed. A dictionary catch-up consumes sequences 0..n-1 before any + // data frame, so after one this test is false and the recycle below -- + // which re-anchors the mapping from scratch -- is the correct path, not + // an avoidable cost. + try { + positionCursorForStart(); + } catch (CatchUpSendException e) { + // Re-anchor's catch-up send failed. fail() here is a fresh, + // non-re-entrant connectLoop entry from the I/O loop body -- + // the same recovery a normal trySendBatch send failure takes. + // Preserve the wrapper on a cap gap so connectLoop's + // isCatchUpCapGap keeps the orphan settle episode (attempt count + // + dwell anchor) alive across the re-anchor recycle; an ordinary + // failure unwraps to the raw cause and restarts the episode, like + // any normal send failure. + fail(isCatchUpCapGap(e) ? e : e.getCause()); + return false; + } + return true; + } + // Frames were already sent on this connection: the linear + // fsnAtZero + wireSeq mapping cannot express the gap the retired + // tail leaves behind. Recycle; swapClient re-anchors the mapping + // at ackedFsn + 1 = the first frame past the tail. One reconnect, + // once per recovery -- and only on the slow path (unacked + // committed frames existed below the tail). + fail(new LineSenderException( + "recycling connection after retiring orphaned deferred tail (wire-seq realignment)")); + return false; } - // Frames were already sent on this connection: the linear - // fsnAtZero + wireSeq mapping cannot express the gap the retired - // tail leaves behind. Recycle; swapClient re-anchors the mapping - // at ackedFsn + 1 = the first frame past the tail. One reconnect, - // once per recovery -- and only on the slow path (unacked - // committed frames existed below the tail). - fail(new LineSenderException( - "recycling connection after retiring orphaned deferred tail (wire-seq realignment)")); - return false; - } - long pub = sendingSegment.publishedOffset(); - if (sendOffset >= pub) { - // Nothing more in the current segment. If it's a sealed segment - // (no longer the live active), advance to the next one. - if (sendingSegment != engine.activeSegment()) { - // The producer can publish the current segment's last frame - // between our first publishedOffset() read and the active - // segment check above. Re-read before leaving the segment, or - // that frame is skipped permanently. - pub = sendingSegment.publishedOffset(); - if (sendOffset >= pub) { - MmapSegment next = advanceSegment(); - if (next != sendingSegment) { - sendingSegment = next; - return true; // let the next iteration try sending + long pub = sendingSegment.publishedOffset(); + if (sendOffset >= pub) { + // Nothing more in the current segment. If it's a sealed segment + // (no longer the live active), advance to the next one. + if (sendingSegment != engine.activeSegment()) { + // The producer can publish the current segment's last frame + // between our first publishedOffset() read and the active + // segment check above. Re-read before leaving the segment, or + // that frame is skipped permanently. + pub = sendingSegment.publishedOffset(); + if (sendOffset >= pub) { + if (batchFrames > 0) { + break; + } + MmapSegment next = advanceSegment(); + if (next != sendingSegment) { + sendingSegment = next; + return true; // let the next iteration try sending + } } + } else { + break; } - } else { + } + // At least the frame header is published; check we have the full frame. + if (sendOffset + MmapSegment.FRAME_HEADER_SIZE > pub) { + break; + } + long base = sendingSegment.address(); + // Frame layout: [u32 crc][u32 payloadLen][payload]. + int payloadLen = Unsafe.getUnsafe().getInt(base + sendOffset + 4); + if (payloadLen < 0) { + if (batchFrames > 0) { + break; + } + fail(new LineSenderException( + "negative payloadLen at offset " + sendOffset + + " in segment baseSeq=" + sendingSegment.baseSeq())); return false; } + long frameEnd = sendOffset + MmapSegment.FRAME_HEADER_SIZE + payloadLen; + if (frameEnd > pub) { + break; // payload not fully published yet + } + long wireLen = WebSocketFrameWriter.headerSize(payloadLen, true) + (long) payloadLen; + if (batchFrames > 0 && batchBytes + wireLen > client.batchCapacity()) { + break; // byte cap: the next call carries on from this frame + } + long frameAddr = base + sendOffset + MmapSegment.FRAME_HEADER_SIZE; + // Torn-dictionary guard. sentDictCount is this loop's model of how many ids the + // CURRENT server has been told about. A frame whose delta starts ABOVE that + // coverage references ids the server was never given, and the server now rejects + // such a frame with STATUS_DICTIONARY_GAP rather than null-padding the hole. That + // rejection is retriable -- a wire recycle re-registers from id 0 -- but this + // frame cannot become sendable without one, and against a server old enough to + // null-pad it would land rows with silently NULL symbols. Fail terminally here, + // where the unreplayable range is still attributable. + // + // The guard, the mirror and the catch-up are ONE mechanism and are all + // ungated (see the constructor). A frame at deltaStart == sentDictCount is + // contiguous and extends the coverage, so a slot whose frames start at id 0 + // replays cleanly with no persisted dictionary at all -- which is exactly why + // the mirror below must keep advancing even when the dictionary is missing. + // A gap here therefore means genuine loss: a host/power crash tore the + // unsynced .symbol-dict below the ids the surviving frames still reference + // (SF is process-crash but not host-crash durable). + int deltaStart = frameDeltaStart(frameAddr, payloadLen); + if (deltaStart > sentDictCount) { + if (batchFrames > 0) { + break; + } + recordFatal(new LineSenderException( + "recovered store-and-forward symbol dictionary is incomplete (likely a host crash): " + + "frame delta start " + deltaStart + " exceeds recovered dictionary size " + + sentDictCount + "; cannot replay without corrupting data -- resend required")); + return false; + } + try { + client.batchAppendBinary(frameAddr, payloadLen); + } catch (Throwable t) { + fail(t); + return false; + } + // The bookkeeping below runs at append time, before the flush: the next + // frame's torn-dictionary guard needs this frame's coverage. If the flush + // fails, fail() re-anchors the wire state at ackedFsn + 1 on a fresh + // client, so an appended-but-unflushed frame is just another unacked + // frame; the persistent dictionary mirror is replayed by the catch-up. + // + // A real ring frame (data or commit) is going out on this connection, + // as opposed to only the dictionary catch-up. onClose / handleServerRejection + // key their poison-strike vs pre-send decision off this, not off nextWireSeq + // (which the catch-up advances). + dataFrameSentThisConnection = true; + if (deltaStart >= 0) { + // Mirror the symbols this frame just registered on the server, so a later + // reconnect can rebuild the whole dictionary and the guard above keeps an + // accurate view of the server's coverage. Idempotent on replay: a frame + // whose delta we already hold advances nothing. + // + // Ungated (see the constructor): this is the ONLY thing that advances + // sentDictCount from the frames themselves, so gating it while leaving the + // guard ungated froze the coverage at 0 and terminal'd a slot that replays + // perfectly from id 0. + accumulateSentDict(frameAddr, payloadLen, deltaStart); + } + sendOffset = frameEnd; + long fsnSent = fsnAtZero + nextWireSeq; + nextWireSeq++; + batchFrames++; + batchBytes += wireLen; + if (replayTargetFsn >= 0) { + batchReplayed++; + if (fsnSent >= replayTargetFsn) { + replayTargetFsn = -1L; // catch-up complete + } + } } - // At least the frame header is published; check we have the full frame. - if (sendOffset + MmapSegment.FRAME_HEADER_SIZE > pub) { - return false; - } - long base = sendingSegment.address(); - // Frame layout: [u32 crc][u32 payloadLen][payload]. - int payloadLen = Unsafe.getUnsafe().getInt(base + sendOffset + 4); - if (payloadLen < 0) { - fail(new LineSenderException( - "negative payloadLen at offset " + sendOffset - + " in segment baseSeq=" + sendingSegment.baseSeq())); - return false; - } - long frameEnd = sendOffset + MmapSegment.FRAME_HEADER_SIZE + payloadLen; - if (frameEnd > pub) { - return false; // payload not fully published yet - } - long frameAddr = base + sendOffset + MmapSegment.FRAME_HEADER_SIZE; - // Torn-dictionary guard. sentDictCount is this loop's model of how many ids the - // CURRENT server has been told about. A frame whose delta starts ABOVE that - // coverage references ids the server was never given, and the server now rejects - // such a frame with STATUS_DICTIONARY_GAP rather than null-padding the hole. That - // rejection is retriable -- a wire recycle re-registers from id 0 -- but this - // frame cannot become sendable without one, and against a server old enough to - // null-pad it would land rows with silently NULL symbols. Fail terminally here, - // where the unreplayable range is still attributable. - // - // The guard, the mirror and the catch-up are ONE mechanism and are all - // ungated (see the constructor). A frame at deltaStart == sentDictCount is - // contiguous and extends the coverage, so a slot whose frames start at id 0 - // replays cleanly with no persisted dictionary at all -- which is exactly why - // the mirror below must keep advancing even when the dictionary is missing. - // A gap here therefore means genuine loss: a host/power crash tore the - // unsynced .symbol-dict below the ids the surviving frames still reference - // (SF is process-crash but not host-crash durable). - int deltaStart = frameDeltaStart(frameAddr, payloadLen); - if (deltaStart > sentDictCount) { - recordFatal(new LineSenderException( - "recovered store-and-forward symbol dictionary is incomplete (likely a host crash): " - + "frame delta start " + deltaStart + " exceeds recovered dictionary size " - + sentDictCount + "; cannot replay without corrupting data -- resend required")); + if (batchFrames == 0) { return false; } try { - client.sendBinary(frameAddr, payloadLen); + client.batchFlush(); } catch (Throwable t) { fail(t); return false; } - // A real ring frame (data or commit) has now gone out on this connection, - // as opposed to only the dictionary catch-up. onClose / handleServerRejection - // key their poison-strike vs pre-send decision off this, not off nextWireSeq - // (which the catch-up advances). - dataFrameSentThisConnection = true; - if (deltaStart >= 0) { - // Mirror the symbols this frame just registered on the server, so a later - // reconnect can rebuild the whole dictionary and the guard above keeps an - // accurate view of the server's coverage. Idempotent on replay: a frame - // whose delta we already hold advances nothing. - // - // Ungated (see the constructor): this is the ONLY thing that advances - // sentDictCount from the frames themselves, so gating it while leaving the - // guard ungated froze the coverage at 0 and terminal'd a slot that replays - // perfectly from id 0. - accumulateSentDict(frameAddr, payloadLen, deltaStart); - } lastFrameOrPingNanos = System.nanoTime(); - sendOffset = frameEnd; - long fsnSent = fsnAtZero + nextWireSeq; - nextWireSeq++; - counters.framesSent.incrementAndGet(); - if (replayTargetFsn >= 0) { - counters.framesReplayed.incrementAndGet(); - if (fsnSent >= replayTargetFsn) { - replayTargetFsn = -1L; // catch-up complete - } + counters.framesSent.addAndGet(batchFrames); + if (batchReplayed > 0) { + counters.framesReplayed.addAndGet(batchReplayed); } return true; } @@ -4288,7 +4338,7 @@ boolean isConnectInFlight() { * current connection. Thrown by {@link #sendDictCatchUp}/{@link * #sendCatchUpChunk} instead of calling {@link #fail}: the catch-up runs * inside {@link #connectLoop} (via {@link #swapClient}) and, on the initial - * connect, inside {@link #start()} / {@link #trySendOne} on the caller + * connect, inside {@link #start()} / {@link #trySendBatch} on the caller * thread. Calling {@code fail()} from there would re-enter {@code * connectLoop} -- corrupting the {@code fsnAtZero}/{@code nextWireSeq} wire * mapping (a subsequent ACK then trims un-acked frames) and growing the @@ -4297,7 +4347,7 @@ boolean isConnectInFlight() { * thread and block {@code Sender} construction indefinitely. Each catch * site instead turns it into ONE non-re-entrant reconnect: {@code * connectLoop}'s own retry catch (swapClient path), a fresh {@code fail()} - * from the I/O loop body (trySendOne path), or dropping the dead client so + * from the I/O loop body (trySendBatch path), or dropping the dead client so * the I/O thread reconnects (start path). A JVM {@code Error} is never * wrapped -- it must stay terminal. The {@code capGap} marker lets the reconnect * loop preserve only consecutive incompatible-cap observations; ordinary catch-up @@ -4607,7 +4657,7 @@ private void handleServerRejection(long wireSeq) { // NACK is for a dictionary catch-up frame -- which occupies the // already-acked wire sequences below replayStart -- not a data frame // this connection sent. (dataFrameSentThisConnection can be true here - // because trySendOne ships the head data frame before tryReceiveAcks + // because trySendBatch ships the head data frame before tryReceiveAcks // reads the catch-up's NACK in the same loop iteration.) Attributing // it a data FSN would key recordHeadRejectionStrike() off a // below-baseline FSN -- negative when replayStart < catchUpFrames -- diff --git a/core/src/test/java/io/questdb/client/test/cutlass/http/client/WebSocketClientTest.java b/core/src/test/java/io/questdb/client/test/cutlass/http/client/WebSocketClientTest.java index 4cf24c6f..26d84d25 100644 --- a/core/src/test/java/io/questdb/client/test/cutlass/http/client/WebSocketClientTest.java +++ b/core/src/test/java/io/questdb/client/test/cutlass/http/client/WebSocketClientTest.java @@ -49,6 +49,63 @@ public class WebSocketClientTest { + /** + * Batched frames go out back to back in ONE send(), each a well-formed + * masked binary frame with an exact-size header (no MAX_HEADER_SIZE gap + * between frames); an empty flush writes nothing. + */ + @Test + public void testBatchFlushSendsAppendedFramesInOneWrite() throws Exception { + assertMemoryLeak(() -> { + CapturingSocket socket = new CapturingSocket(); + // 7: 7-bit length, 200: 16-bit length, 0: empty payload + int[] lens = {7, 200, 0}; + long payload = Unsafe.malloc(200, MemoryTag.NATIVE_DEFAULT); + try (CapturingWebSocketClient client = new CapturingWebSocketClient(socket)) { + setUpgradedTrue(client); + for (int i = 0; i < 200; i++) { + Unsafe.getUnsafe().putByte(payload + i, (byte) i); + } + + client.batchFlush(); + Assert.assertEquals("an empty flush must not write", 0, socket.sendCalls); + + for (int len : lens) { + client.batchAppendBinary(payload, len); + } + Assert.assertEquals("appending must not write", 0, socket.sendCalls); + client.batchFlush(); + Assert.assertEquals("the whole batch must be one write", 1, socket.sendCalls); + + byte[] wire = socket.sent; + int pos = 0; + for (int len : lens) { + Assert.assertEquals("FIN + BINARY", (byte) 0x82, wire[pos]); + int maskPos; + if (len <= 125) { + Assert.assertEquals((byte) (0x80 | len), wire[pos + 1]); + maskPos = pos + 2; + } else { + Assert.assertEquals((byte) (0x80 | 126), wire[pos + 1]); + Assert.assertEquals(len, ((wire[pos + 2] & 0xFF) << 8) | (wire[pos + 3] & 0xFF)); + maskPos = pos + 4; + } + int payloadPos = maskPos + 4; + for (int i = 0; i < len; i++) { + Assert.assertEquals((byte) i, (byte) (wire[payloadPos + i] ^ wire[maskPos + (i & 3)])); + } + pos = payloadPos + len; + } + Assert.assertEquals("no bytes beyond the 3 frames", wire.length, pos); + + client.batchFlush(); + Assert.assertEquals("a flushed batch is empty", 1, socket.sendCalls); + } finally { + Unsafe.free(payload, 200, MemoryTag.NATIVE_DEFAULT); + } + }); + } + /** * close() frees native memory (recv/fragment buffers, send buffers), so * its guard must be a CAS, not a volatile check-then-act: two concurrent diff --git a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/BackgroundDrainerInterruptedTeardownTest.java b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/BackgroundDrainerInterruptedTeardownTest.java index d3ec81ee..36d38867 100644 --- a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/BackgroundDrainerInterruptedTeardownTest.java +++ b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/BackgroundDrainerInterruptedTeardownTest.java @@ -267,7 +267,7 @@ private static final class StubWebSocketClient extends TestWebSocketClient { } @Override - public void sendBinary(long dataPtr, int length) { + public void batchAppendBinary(long dataPtr, int length) { throw new IllegalStateException("stub: wire down"); } diff --git a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoopBlockedSendCloseTest.java b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoopBlockedSendCloseTest.java index 1d5f16ec..3f799663 100644 --- a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoopBlockedSendCloseTest.java +++ b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoopBlockedSendCloseTest.java @@ -231,7 +231,12 @@ public void closeTraffic() { } @Override - public void sendBinary(long dataPtr, int length, int timeout) { + public void batchAppendBinary(long dataPtr, int length) { + // the frame goes out (and blocks) in batchFlush + } + + @Override + public void batchFlush() { sendThread.set(Thread.currentThread()); sendEntered.countDown(); try { diff --git a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoopCatchUpAlignmentTest.java b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoopCatchUpAlignmentTest.java index 5717aa17..b433a9e5 100644 --- a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoopCatchUpAlignmentTest.java +++ b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoopCatchUpAlignmentTest.java @@ -1107,7 +1107,7 @@ private static void appendFrames(CursorSendEngine engine, int count) { * varints. deltaStart 0 is the self-sufficient full-dict frame a degraded producer * ships; deltaStart > 0 is a delta frame that depends on the ids below it. */ - private static void appendDeltaDictFrame(CursorSendEngine engine, int deltaStart, char... symbols) { + static void appendDeltaDictFrame(CursorSendEngine engine, int deltaStart, char... symbols) { int size = QwpConstants.HEADER_SIZE + 2 + symbols.length * 2; long buf = Unsafe.malloc(size, MemoryTag.NATIVE_DEFAULT); try { @@ -1682,7 +1682,7 @@ public int getServerQwpVersion() { } @Override - public void sendBinary(long dataPtr, int length) { + public void batchAppendBinary(long dataPtr, int length) { recordSend(); capture(dataPtr, length, 0L, 0); } diff --git a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoopForegroundReconnectPolicyTest.java b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoopForegroundReconnectPolicyTest.java index 54420fe5..be10cbcf 100644 --- a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoopForegroundReconnectPolicyTest.java +++ b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoopForegroundReconnectPolicyTest.java @@ -491,7 +491,7 @@ public int getServerQwpVersion() { } @Override - public void sendBinary(long dataPtr, int length) { + public void batchAppendBinary(long dataPtr, int length) { throw new RuntimeException("transient wire failure during catch-up"); } diff --git a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoopIdleWaitTest.java b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoopIdleWaitTest.java index 22d30a0e..78c98506 100644 --- a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoopIdleWaitTest.java +++ b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoopIdleWaitTest.java @@ -217,6 +217,43 @@ public void testReplayAfterPingFailureIsNotStrandedByIdleWait() throws Exception }); } + /** + * A batch never outgrows the send buffer: it ends before the frame that + * would not fit, so the ACK side gets a turn per buffer's worth of bytes + * and the buffer is never grown towards its maximum by a backlog. + */ + @Test + public void testBurstIsSplitAtSendBufferCapacity() throws Exception { + TestUtils.assertMemoryLeak(() -> { + try (CursorSendEngine engine = new CursorSendEngine(tmpDir, 1 << 20)) { + BlockingDoorbellClient client = new BlockingDoorbellClient(); + final int frames = 16; + final int payloadLen = 8192; + final int wireLen = payloadLen + 8; // 2-byte header + 16-bit length + 4-byte mask key + int framesPerWrite = client.batchCapacity() / wireLen; + assertTrue("test needs several frames per write", framesPerWrite > 1 && framesPerWrite < frames); + publishBurstToHeldLoop(engine, client, frames, payloadLen); + assertEquals("batches must be cut at the send buffer's capacity", + (frames + framesPerWrite - 1) / framesPerWrite, client.writes.get()); + } + }); + } + + /** + * Frames already published when the loop reaches them share one socket + * write, instead of costing a send() (plus an EAGAIN recv()) each. + */ + @Test + public void testBurstOfPublishedFramesGoesOutInOneWrite() throws Exception { + TestUtils.assertMemoryLeak(() -> { + try (CursorSendEngine engine = new CursorSendEngine(tmpDir, 16384)) { + BlockingDoorbellClient client = new BlockingDoorbellClient(); + publishBurstToHeldLoop(engine, client, 8, 16); + assertEquals("8 published frames must share one write", 1, client.writes.get()); + } + }); + } + /** * The publish hook rings the doorbell only while the I/O thread is parked * (or committing to park). A loop that is busy -- here, inside the @@ -265,7 +302,7 @@ public void testBusyLoopIsNotWokenPerPublish() throws Exception { /** * The idle decision re-checks the engine's publish stamp after committing - * to park: a publish that lands after trySendOne found nothing but before + * to park: a publish that lands after trySendBatch found nothing but before * ioParked is set rings no doorbell, so without the re-check the loop * would block on an untimed wait with the frame stranded behind it. The * transport's receive, which runs inside that window, is gated so the @@ -292,7 +329,7 @@ public void testPublishInsideIdleDecisionWindowIsNotStranded() throws Exception assertTrue("loop never reached the idle-decision window", client.receiveHeld.await(5, TimeUnit.SECONDS)); assertEquals(1, client.sentFrames.get()); - // The loop is between trySendOne (found nothing) and the + // The loop is between trySendBatch (found nothing) and the // park decision: this publish rings no doorbell. int wakesBefore = client.wakes.get(); appendFrame(engine); @@ -325,6 +362,50 @@ private static void appendFrame(CursorSendEngine engine) { } } + private static void appendZeroFrame(CursorSendEngine engine, int len) { + long buf = Unsafe.calloc(len, MemoryTag.NATIVE_DEFAULT); + try { + engine.appendBlocking(buf, len); + } finally { + Unsafe.free(buf, len, MemoryTag.NATIVE_DEFAULT); + } + } + + /** + * Holds the loop in its first receive, publishes the frames, releases it + * and waits until the loop has counted them all as sent. + */ + private static void publishBurstToHeldLoop( + CursorSendEngine engine, BlockingDoorbellClient client, int frames, int payloadLen + ) throws Exception { + CountDownLatch gate = new CountDownLatch(1); + client.receiveGate = gate; + CursorWebSocketSendLoop loop = new CursorWebSocketSendLoop( + client, engine, 0L, + BlockingDoorbellClient::new, + 100L, 1_000L, + DurableAckTiers.NONE, + CursorWebSocketSendLoop.DEFAULT_DURABLE_ACK_KEEPALIVE_INTERVAL_MILLIS, + CursorWebSocketSendLoop.DEFAULT_MAX_HEAD_FRAME_REJECTIONS); + try { + loop.start(); + assertTrue("loop never entered the held receive", client.receiveHeld.await(5, TimeUnit.SECONDS)); + for (int i = 0; i < frames; i++) { + appendZeroFrame(engine, payloadLen); + } + gate.countDown(); + long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(5); + while (loop.getTotalFramesSent() < frames && System.nanoTime() < deadline) { + Thread.sleep(1); + } + assertEquals("not every published frame was sent", frames, loop.getTotalFramesSent()); + assertEquals(frames, client.sentFrames.get()); + } finally { + gate.countDown(); + loop.close(); + } + } + /** * A transport whose idle wait really blocks (no bounded park), like the * real one, with a stateful doorbell. @@ -335,6 +416,7 @@ private static final class BlockingDoorbellClient extends WebSocketClient { final AtomicInteger waits = new AtomicInteger(); final AtomicInteger pingAttempts = new AtomicInteger(); final AtomicInteger wakes = new AtomicInteger(); + final AtomicInteger writes = new AtomicInteger(); // Counted down once the receive below is held at receiveGate. final CountDownLatch receiveHeld = new CountDownLatch(1); volatile boolean failFirstSend; @@ -345,7 +427,7 @@ private static final class BlockingDoorbellClient extends WebSocketClient { // One-shot: the receive that takes the gate blocks on it (5 s cap). // Plain: the very first receive. gateInIdleWindow: the second // receive after the first send, i.e. the one that runs after - // trySendOne found nothing and before the park decision. + // trySendBatch found nothing and before the park decision. volatile CountDownLatch receiveGate; volatile boolean gateInIdleWindow; private boolean okDelivered; @@ -381,7 +463,7 @@ public void wake() { } @Override - public void sendBinary(long dataPtr, int length) { + public void batchAppendBinary(long dataPtr, int length) { if (failFirstSend) { failFirstSend = false; throw new HttpClientException("peer disconnect [errno=104]"); @@ -390,6 +472,19 @@ public void sendBinary(long dataPtr, int length) { receivesSinceSend = 0; } + // One socket write per flush, like the real client. + @Override + public void batchFlush() { + writes.incrementAndGet(); + } + + // The one-frame-per-write path: one frame, one write. + @Override + public void sendBinary(long dataPtr, int length) { + batchAppendBinary(dataPtr, length); + batchFlush(); + } + @Override public void sendPing() { pingAttempts.incrementAndGet(); diff --git a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoopOrphanTailTest.java b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoopOrphanTailTest.java index 22c4a397..aa6d9cab 100644 --- a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoopOrphanTailTest.java +++ b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoopOrphanTailTest.java @@ -192,7 +192,7 @@ public void testFastPathRetiresWholeDeferredLogBeforeAnySend() throws Exception * dictionary, so nothing pinned that adding a catch-up -- which consumes wire * sequences before any data frame -- leaves the start()-time retirement intact. *

- * This deliberately does NOT pin trySendOne's in-place re-anchor arm: both connection + * This deliberately does NOT pin trySendBatch's in-place re-anchor arm: both connection * setup sites call tryRetireOrphanTail first, so that arm is only reached when frames * below the tail still needed acks -- which means they were sent on this connection. * Verified by reverting its guard: this test still passes. @@ -362,7 +362,7 @@ public void testRecoveredMaxSymbolIdSpansSealedSegments() throws Exception { @Test public void testRecoveredMaxSymbolIdExcludesOrphanTailFrames() throws Exception { // recoveredMaxSymbolId must reflect only COMMITTED (transmitted) frames, not - // the aborted orphan-tail frames trySendOne retires without ever sending. A + // the aborted orphan-tail frames trySendBatch retires without ever sending. A // host crash that tears the persisted dictionary down to the committed ids // while an orphan-tail frame introduced a HIGHER id must NOT over-reject the // resume: the producer never reuses an orphan id on the wire (the tail retires @@ -648,7 +648,7 @@ public void testSelfSufficientFrameCannotHideUnackedRecoveryGap() throws Excepti /** * In-memory transport emulating a healthy server: counts sends and * answers every received frame with a cumulative empty-table STATUS_OK - * for the highest wire seq seen. sendBinary and tryReceiveFrame both run + * for the highest wire seq seen. batchAppendBinary and tryReceiveFrame both run * on the I/O thread; sentCount is volatile for test-thread assertions. */ private static final class AckingClient extends TestWebSocketClient { @@ -660,7 +660,7 @@ private static final class AckingClient extends TestWebSocketClient { } @Override - public void sendBinary(long dataPtr, int length) { + public void batchAppendBinary(long dataPtr, int length) { sentCount++; } diff --git a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoopPoisonFrameTest.java b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoopPoisonFrameTest.java index ae083b8f..52d04d49 100644 --- a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoopPoisonFrameTest.java +++ b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoopPoisonFrameTest.java @@ -893,7 +893,7 @@ private static final class NackingClient extends TestWebSocketClient { } @Override - public void sendBinary(long dataPtr, int length) { + public void batchAppendBinary(long dataPtr, int length) { sentCount++; } @@ -940,7 +940,7 @@ private static final class ClosingMiddleboxClient extends TestWebSocketClient { } @Override - public void sendBinary(long dataPtr, int length) { + public void batchAppendBinary(long dataPtr, int length) { sentCount++; } @@ -978,7 +978,7 @@ private static final class OrderlyClosingClient extends TestWebSocketClient { } @Override - public void sendBinary(long dataPtr, int length) { + public void batchAppendBinary(long dataPtr, int length) { } @Override @@ -1016,7 +1016,7 @@ private static final class PreSendClosingClient extends TestWebSocketClient { } @Override - public void sendBinary(long dataPtr, int length) { + public void batchAppendBinary(long dataPtr, int length) { } @Override diff --git a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoopRotationRaceTest.java b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoopRotationRaceTest.java index b21a018b..0c0e8b57 100644 --- a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoopRotationRaceTest.java +++ b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoopRotationRaceTest.java @@ -42,7 +42,7 @@ /** * Probabilistic regression test for the send-loop rotation race fixed in - * {@code CursorWebSocketSendLoop.trySendOne()} (PR #41): the loop could observe + * {@code CursorWebSocketSendLoop.trySendBatch()} (PR #41): the loop could observe * a segment rotation without observing the final frame published into that * segment, and advance past it — silently dropping the frame. *

@@ -85,7 +85,7 @@ public void tailFrameSurvivesSegmentRotation() throws Exception { SegmentManager manager = new SegmentManager( SEGMENT_BYTES, new SegmentBudget(SegmentManager.UNLIMITED_TOTAL_BYTES)); manager.start(); - // sfDir == null => memory-only segments (no files/fds); same trySendOne() path. + // sfDir == null => memory-only segments (no files/fds); same trySendBatch() path. CursorSendEngine engine = new CursorSendEngine(null, SEGMENT_BYTES, manager); RecordingClient client = new RecordingClient(n); CursorWebSocketSendLoop loop = new CursorWebSocketSendLoop( @@ -137,7 +137,7 @@ private static void assertContiguous(RecordingClient client, int expected) { } /** - * In-memory transport, and nothing more: {@code sendBinary} captures the fsn + * In-memory transport, and nothing more: {@code batchAppendBinary} captures the fsn * carried by each frame the loop ships. The other callbacks are inert — this * test never receives ACKs and never reconnects. Only {@code sentCount} is * read cross-thread (by the test, for progress) and so is volatile. @@ -153,7 +153,7 @@ private static final class RecordingClient extends TestWebSocketClient { } @Override - public void sendBinary(long dataPtr, int length) { + public void batchAppendBinary(long dataPtr, int length) { if (recordedCount < recorded.length) { recorded[recordedCount++] = Unsafe.getUnsafe().getLong(dataPtr); } diff --git a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoopTornDictGuardTest.java b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoopTornDictGuardTest.java index c47c06cd..343a7036 100644 --- a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoopTornDictGuardTest.java +++ b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoopTornDictGuardTest.java @@ -52,7 +52,7 @@ /** * White-box coverage of the send loop's torn-dictionary guard - * ({@code trySendOne}: {@code deltaStart > sentDictCount}) firing on a genuine gap. + * ({@code trySendBatch}: {@code deltaStart > sentDictCount}) firing on a genuine gap. *

* That guard is the drainer path's SOLE defense: an orphan drainer adopts a slot * without running {@code Sender.build()}'s seed-time guard, so on a slot whose @@ -84,6 +84,48 @@ public void tearDown() { TestUtils.removeTmpDirRec(sfDir); } + /** + * A gapped frame behind good frames ends the batch BEFORE it: the good frames + * are flushed first, and the guard fires on the next pass with nothing + * pending -- as when each frame had its own send. Firing mid-batch would latch + * the terminal with the good frames appended but never written. + */ + @Test + public void testGuardFiresOnlyAfterFramesAheadOfTheGapAreFlushed() throws Exception { + assertMemoryLeak(() -> { + CountingClient client = new CountingClient(); + try (CursorSendEngine engine = new CursorSendEngine(null, 16_384)) { + CursorWebSocketSendLoopCatchUpAlignmentTest.appendDeltaDictFrame(engine, 0, 'a'); + CursorWebSocketSendLoopCatchUpAlignmentTest.appendDeltaDictFrame(engine, 1, 'b'); + // the server only ever learns ids 0..1, so a delta starting at 5 is a gap + CursorWebSocketSendLoopCatchUpAlignmentTest.appendDeltaDictFrame(engine, 5, 'f'); + CursorWebSocketSendLoop loop = newLoop(engine, client); + try { + loop.setRunningForTest(true); + loop.positionCursorForStartForTest(); + + Assert.assertTrue("the frames ahead of the gap must be sent", loop.trySendBatchForTest()); + Assert.assertEquals(2, client.framesSent); + Assert.assertEquals("the frames ahead of the gap must be flushed", 1, client.flushes); + Assert.assertEquals(2, loop.getTotalFramesSent()); + loop.checkError(); // no terminal yet + + Assert.assertFalse("the gapped frame must be refused", loop.trySendBatchForTest()); + Assert.assertEquals("no frame may reach the server through a gap", 2, client.framesSent); + try { + loop.checkError(); + Assert.fail("the guard must latch a terminal error"); + } catch (LineSenderException e) { + Assert.assertTrue("unexpected terminal: " + e.getMessage(), + e.getMessage().contains("incomplete")); + } + } finally { + loop.close(); + } + } + }); + } + @Test public void testGuardFiresOnGenuineGapAndShipsNoFrame() throws Exception { assertMemoryLeak(() -> { @@ -104,7 +146,7 @@ public void testGuardFiresOnGenuineGapAndShipsNoFrame() throws Exception { // before its first send (no catch-up: the mirror is empty). loop.positionCursorForStartForTest(); - boolean sent = loop.trySendOneForTest(); + boolean sent = loop.trySendBatchForTest(); Assert.assertFalse("the torn-dict guard must refuse to send the gapped frame", sent); Assert.assertEquals("no frame may reach the server through a gap", @@ -138,7 +180,7 @@ private static int countSegmentFiles(Path dir) { } // Constructs a recovery send loop that is never started -- the test drives - // positionCursorForStart + trySendOne directly. The reconnect factory throws + // positionCursorForStart + trySendBatch directly. The reconnect factory throws // because no reconnect is expected before the guard latches its terminal. private CursorWebSocketSendLoop newLoop(CursorSendEngine engine, WebSocketClient client) { return new CursorWebSocketSendLoop( @@ -185,8 +227,9 @@ private void writeAndTearGappedSlot() throws Exception { } // Frame-counting stub transport: completes no real I/O. If the guard ever lets a - // gapped frame through, sendBinary bumps framesSent and the test fails. + // gapped frame through, batchAppendBinary bumps framesSent and the test fails. private static final class CountingClient extends TestWebSocketClient { + private int flushes; private int framesSent; CountingClient() { @@ -204,10 +247,15 @@ public int getServerQwpVersion() { } @Override - public void sendBinary(long dataPtr, int length) { + public void batchAppendBinary(long dataPtr, int length) { framesSent++; } + @Override + public void batchFlush() { + flushes++; + } + @Override protected void ioWait(int timeout, int op) { } From 7bdf94a1b564373aa7dd040454dadff166dc2664 Mon Sep 17 00:00:00 2001 From: Jaromir Hamala Date: Wed, 4 Nov 2026 16:27:49 +0100 Subject: [PATCH 03/13] perf(ws): compact the receive buffer lazily, once per recv tryParseFrame compacted the unread tail after every parsed frame, so a burst of N small frames in one recv cost N memmoves of a shrinking tail (quadratic in the burst; 82% of the I/O thread's CPU under per-frame server acks). Rewind the cursors only when the buffer is fully consumed and compact a partial tail once, right before the next recv and before the grow check, so the buffer state at every recv is unchanged. --- .../cutlass/http/client/WebSocketClient.java | 11 +++- .../cutlass/qwp/client/QwpEgressIoThread.java | 10 ++-- .../http/client/WebSocketClientTest.java | 60 +++++++++++++++++++ 3 files changed, 74 insertions(+), 7 deletions(-) diff --git a/core/src/main/java/io/questdb/client/cutlass/http/client/WebSocketClient.java b/core/src/main/java/io/questdb/client/cutlass/http/client/WebSocketClient.java index dd6ca2b1..d314c9d2 100644 --- a/core/src/main/java/io/questdb/client/cutlass/http/client/WebSocketClient.java +++ b/core/src/main/java/io/questdb/client/cutlass/http/client/WebSocketClient.java @@ -517,6 +517,7 @@ public boolean receiveFrame(WebSocketFrameHandler handler, int timeout) { } // Ensure buffer has space + compactRecvBuffer(); if (recvPos >= recvBufSize - 1024) { growRecvBuffer(); } @@ -702,6 +703,7 @@ public boolean tryReceiveFrame(WebSocketFrameHandler handler) { } // Try one non-blocking recv + compactRecvBuffer(); if (recvPos >= recvBufSize - 1024) { growRecvBuffer(); } @@ -1453,8 +1455,13 @@ private int tryParseFrame(WebSocketFrameHandler handler) { // Advance read position recvReadPos += consumed; - // Compact buffer if needed - compactRecvBuffer(); + // Rewind once drained; a partial tail stays put until the next + // recv compacts it. Compacting here, per frame, memmoved the rest + // of a burst once per frame: quadratic in the burst. + if (recvReadPos == recvPos) { + recvReadPos = 0; + recvPos = 0; + } return PARSE_OK; } diff --git a/core/src/main/java/io/questdb/client/cutlass/qwp/client/QwpEgressIoThread.java b/core/src/main/java/io/questdb/client/cutlass/qwp/client/QwpEgressIoThread.java index ecfa91f7..153b57e9 100644 --- a/core/src/main/java/io/questdb/client/cutlass/qwp/client/QwpEgressIoThread.java +++ b/core/src/main/java/io/questdb/client/cutlass/qwp/client/QwpEgressIoThread.java @@ -638,11 +638,11 @@ private void handleResultBatch(long payloadPtr, int payloadLen) { // releaseBuffer offers the token that unblocks this take. // // Wait uninterruptibly: if close() interrupts us here we MUST NOT - // unwind back to tryParseFrame, because tryParseFrame's tail - // advances recvReadPos and runs compactRecvBuffer (Vect.memmove) - // over the bytes the user's column pointers still alias inside - // recv-buf. A premature unwind on interrupt is the close()-during- - // onBatch UAF: the user thread's onBatch reads garbage mid-call. + // unwind back to tryParseFrame, because the next receive's + // compactRecvBuffer (Vect.memmove) and recv write over the bytes the + // user's column pointers still alias inside recv-buf. A premature + // unwind on interrupt is the close()-during-onBatch UAF: the user + // thread's onBatch reads garbage mid-call. // Stay parked; close()'s 5-second join timeout will abandon the // I/O thread (the documented timeout-leak path), but only AFTER // the user's onBatch has finished -- no UAF. diff --git a/core/src/test/java/io/questdb/client/test/cutlass/http/client/WebSocketClientTest.java b/core/src/test/java/io/questdb/client/test/cutlass/http/client/WebSocketClientTest.java index 26d84d25..25d0aa4e 100644 --- a/core/src/test/java/io/questdb/client/test/cutlass/http/client/WebSocketClientTest.java +++ b/core/src/test/java/io/questdb/client/test/cutlass/http/client/WebSocketClientTest.java @@ -189,6 +189,66 @@ public void testExtractMaxBatchSizeParsesPositive() throws Exception { Assert.assertEquals(16 * 1024 * 1024, invokeExtractMaxBatchSize(response)); } + /** + * One recv delivering a burst of small frames (the shape of a server's + * cumulative ACKs) must be parsed in place: compacting the unread tail + * after every frame memmoved the rest of the burst once per frame. The + * burst fills the 64 KiB buffer and ends mid-frame, so the next recv must + * compact that tail BEFORE the grow check -- a tail left at the buffer end + * would read as "full" and grow the buffer for no reason. + */ + @Test + public void testFrameBurstIsParsedInPlace() throws Exception { + assertMemoryLeak(() -> { + final int frameLen = 13; // FIN|BINARY, len=11 + final int frameCount = 5100; // > 64 KiB, so the first recv ends mid-frame + byte[] data = new byte[frameLen * frameCount]; + for (int i = 0; i < frameCount; i++) { + data[i * frameLen] = (byte) 0x82; + data[i * frameLen + 1] = 11; + data[i * frameLen + 2] = (byte) i; + data[i * frameLen + 3] = (byte) (i >> 8); + } + try (FrameFeedWebSocketClient client = new FrameFeedWebSocketClient(data)) { + setUpgradedTrue(client); + final int bufSize = getIntField(client, "recvBufSize"); + final long[] payload = {0, 0}; // first, last + final int[] frames = {0}; + WebSocketFrameHandler handler = new WebSocketFrameHandler() { + @Override + public void onBinaryMessage(long payloadPtr, int payloadLen) { + Assert.assertEquals(11, payloadLen); + int tag = (Unsafe.getUnsafe().getByte(payloadPtr) & 0xFF) + | (Unsafe.getUnsafe().getByte(payloadPtr + 1) & 0xFF) << 8; + Assert.assertEquals(frames[0], tag); + if (frames[0]++ == 0) { + payload[0] = payloadPtr; + } + payload[1] = payloadPtr; + } + + @Override + public void onClose(int code, String reason) { + Assert.fail("unexpected close frame"); + } + }; + + int firstRecvFrames = bufSize / frameLen; + for (int i = 0; i < firstRecvFrames; i++) { + Assert.assertTrue(client.tryReceiveFrame(handler)); + Assert.assertEquals("frame " + i + " must be parsed where recv put it", + (long) i * frameLen, payload[1] - payload[0]); + } + for (int i = firstRecvFrames; i < frameCount; i++) { + Assert.assertTrue(client.tryReceiveFrame(handler)); + } + Assert.assertEquals(frameCount, frames[0]); + Assert.assertEquals("partial tail must not grow the recv buffer", + bufSize, getIntField(client, "recvBufSize")); + } + }); + } + /** * A frame handler may close() the client from inside its callback: * CursorWebSocketSendLoop's NACK-recycle path (handleServerRejection / From 2732e4a1d8b9d7a2ea374bf9eeffbcc5b477e0b1 Mon Sep 17 00:00:00 2001 From: Jaromir Hamala Date: Wed, 4 Nov 2026 17:38:54 +0100 Subject: [PATCH 04/13] perf(qwp): cap send batches at a fixed 64 KiB, not the grown buffer size The batch cap followed the send buffer's live capacity, which only ever grows, so one large frame raised the ACK-fairness window for the rest of the connection. Use the default send buffer size as a constant cap; a frame larger than it still goes out alone. --- .../cutlass/http/client/WebSocketClient.java | 10 +-- .../sf/cursor/CursorWebSocketSendLoop.java | 11 ++-- .../CursorWebSocketSendLoopIdleWaitTest.java | 62 +++++++++++++++---- 3 files changed, 62 insertions(+), 21 deletions(-) diff --git a/core/src/main/java/io/questdb/client/cutlass/http/client/WebSocketClient.java b/core/src/main/java/io/questdb/client/cutlass/http/client/WebSocketClient.java index d314c9d2..938d5175 100644 --- a/core/src/main/java/io/questdb/client/cutlass/http/client/WebSocketClient.java +++ b/core/src/main/java/io/questdb/client/cutlass/http/client/WebSocketClient.java @@ -245,7 +245,7 @@ public WebSocketClient(HttpClientConfiguration configuration, SocketFactory sock * Appends one binary frame to the send batch, back to back with any frame * already appended. Nothing reaches the wire until {@link #batchFlush()}. * Grows the send buffer if the frame does not fit; a caller bounding the - * batch checks {@link #batchCapacity()} first. The batch shares the send + * batch checks {@link #batchByteCap()} first. The batch shares the send * buffer with {@link #sendBinary}, so do not interleave the two. */ public void batchAppendBinary(long dataPtr, int length) { @@ -254,10 +254,12 @@ public void batchAppendBinary(long dataPtr, int length) { } /** - * Bytes the send batch holds without growing the send buffer. + * Byte budget for one send batch: the default send buffer size, fixed for + * the connection. It does not follow the buffer's capacity, which only + * grows, so one large frame cannot widen every later batch. */ - public int batchCapacity() { - return sendBuffer.getCapacity(); + public int batchByteCap() { + return DEFAULT_SEND_BUFFER_SIZE; } /** 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 index 8b77d65f..47199487 100644 --- 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 @@ -3879,9 +3879,10 @@ private boolean tryReceiveAcks() { /** * Returns true if at least one frame was sent (caller skips the park). * Sends every fully published frame of the current segment in one write, - * up to the send buffer's capacity, so the ACK side still gets a turn - * every buffer's worth of bytes and the buffer never grows mid-batch (a - * single frame larger than the buffer goes alone and grows it). + * up to the client's fixed batch byte cap (the default send buffer size, + * not the buffer's live capacity), so the ACK side still gets a turn every + * cap's worth of bytes and the buffer never grows mid-batch (a single + * frame larger than the cap goes alone and grows the buffer). *

* Every per-frame check runs per frame, in the old order. A frame that is * not a plain send (orphan tail, corrupt length, torn dictionary, segment @@ -3992,8 +3993,8 @@ private boolean trySendBatch() { break; // payload not fully published yet } long wireLen = WebSocketFrameWriter.headerSize(payloadLen, true) + (long) payloadLen; - if (batchFrames > 0 && batchBytes + wireLen > client.batchCapacity()) { - break; // byte cap: the next call carries on from this frame + if (batchFrames > 0 && batchBytes + wireLen > client.batchByteCap()) { + break; // fixed byte cap: the next call carries on from this frame } long frameAddr = base + sendOffset + MmapSegment.FRAME_HEADER_SIZE; // Torn-dictionary guard. sentDictCount is this loop's model of how many ids the diff --git a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoopIdleWaitTest.java b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoopIdleWaitTest.java index 78c98506..f55ad33f 100644 --- a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoopIdleWaitTest.java +++ b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoopIdleWaitTest.java @@ -218,27 +218,53 @@ public void testReplayAfterPingFailureIsNotStrandedByIdleWait() throws Exception } /** - * A batch never outgrows the send buffer: it ends before the frame that - * would not fit, so the ACK side gets a turn per buffer's worth of bytes + * A batch never outgrows the byte cap: it ends before the frame that + * would not fit, so the ACK side gets a turn per cap's worth of bytes * and the buffer is never grown towards its maximum by a backlog. */ @Test - public void testBurstIsSplitAtSendBufferCapacity() throws Exception { + public void testBurstIsSplitAtBatchByteCap() throws Exception { TestUtils.assertMemoryLeak(() -> { try (CursorSendEngine engine = new CursorSendEngine(tmpDir, 1 << 20)) { BlockingDoorbellClient client = new BlockingDoorbellClient(); final int frames = 16; final int payloadLen = 8192; final int wireLen = payloadLen + 8; // 2-byte header + 16-bit length + 4-byte mask key - int framesPerWrite = client.batchCapacity() / wireLen; + int framesPerWrite = client.batchByteCap() / wireLen; assertTrue("test needs several frames per write", framesPerWrite > 1 && framesPerWrite < frames); - publishBurstToHeldLoop(engine, client, frames, payloadLen); - assertEquals("batches must be cut at the send buffer's capacity", + publishBurstToHeldLoop(engine, client, 0, frames, payloadLen); + assertEquals("batches must be cut at the batch byte cap", (frames + framesPerWrite - 1) / framesPerWrite, client.writes.get()); } }); } + /** + * The byte cap is fixed, not the send buffer's live capacity: a frame + * larger than the cap goes out alone and grows the buffer, and the burst + * behind it is still cut at the cap. A cap that followed the grown buffer + * would let the burst ride along in one or two writes, widening the + * ACK-fairness window for the rest of the connection. + */ + @Test + public void testBatchByteCapDoesNotFollowSendBufferGrowth() throws Exception { + TestUtils.assertMemoryLeak(() -> { + try (CursorSendEngine engine = new CursorSendEngine(tmpDir, 1 << 20)) { + BlockingDoorbellClient client = new BlockingDoorbellClient(); + final int leadPayloadLen = 200 * 1024; + final int frames = 32; + final int payloadLen = 8192 - 8; // wire frame of exactly 8 KiB + int framesPerWrite = client.batchByteCap() / (payloadLen + 8); + assertEquals("test assumes eight 8 KiB frames per 64 KiB cap", 8, framesPerWrite); + publishBurstToHeldLoop(engine, client, leadPayloadLen, frames, payloadLen); + assertTrue("the lead frame must have grown the send buffer past the cap", + client.maxBufferCapacity > client.batchByteCap()); + assertEquals("lead frame alone, then the burst cut at the fixed cap", + 1 + frames / framesPerWrite, client.writes.get()); + } + }); + } + /** * Frames already published when the loop reaches them share one socket * write, instead of costing a send() (plus an EAGAIN recv()) each. @@ -248,7 +274,7 @@ public void testBurstOfPublishedFramesGoesOutInOneWrite() throws Exception { TestUtils.assertMemoryLeak(() -> { try (CursorSendEngine engine = new CursorSendEngine(tmpDir, 16384)) { BlockingDoorbellClient client = new BlockingDoorbellClient(); - publishBurstToHeldLoop(engine, client, 8, 16); + publishBurstToHeldLoop(engine, client, 0, 8, 16); assertEquals("8 published frames must share one write", 1, client.writes.get()); } }); @@ -372,11 +398,12 @@ private static void appendZeroFrame(CursorSendEngine engine, int len) { } /** - * Holds the loop in its first receive, publishes the frames, releases it + * Holds the loop in its first receive, publishes the frames (behind one + * lead frame of leadPayloadLen bytes when that is positive), releases it * and waits until the loop has counted them all as sent. */ private static void publishBurstToHeldLoop( - CursorSendEngine engine, BlockingDoorbellClient client, int frames, int payloadLen + CursorSendEngine engine, BlockingDoorbellClient client, int leadPayloadLen, int frames, int payloadLen ) throws Exception { CountDownLatch gate = new CountDownLatch(1); client.receiveGate = gate; @@ -390,16 +417,21 @@ private static void publishBurstToHeldLoop( try { loop.start(); assertTrue("loop never entered the held receive", client.receiveHeld.await(5, TimeUnit.SECONDS)); + int total = frames; + if (leadPayloadLen > 0) { + appendZeroFrame(engine, leadPayloadLen); + total++; + } for (int i = 0; i < frames; i++) { appendZeroFrame(engine, payloadLen); } gate.countDown(); long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(5); - while (loop.getTotalFramesSent() < frames && System.nanoTime() < deadline) { + while (loop.getTotalFramesSent() < total && System.nanoTime() < deadline) { Thread.sleep(1); } - assertEquals("not every published frame was sent", frames, loop.getTotalFramesSent()); - assertEquals(frames, client.sentFrames.get()); + assertEquals("not every published frame was sent", total, loop.getTotalFramesSent()); + assertEquals(total, client.sentFrames.get()); } finally { gate.countDown(); loop.close(); @@ -430,6 +462,8 @@ private static final class BlockingDoorbellClient extends WebSocketClient { // trySendBatch found nothing and before the park decision. volatile CountDownLatch receiveGate; volatile boolean gateInIdleWindow; + // Largest send buffer capacity seen at a flush. + volatile int maxBufferCapacity; private boolean okDelivered; private int receivesSinceSend; private final Semaphore doorbell = new Semaphore(0); @@ -468,6 +502,8 @@ public void batchAppendBinary(long dataPtr, int length) { failFirstSend = false; throw new HttpClientException("peer disconnect [errno=104]"); } + // Through the real send buffer, so a large frame grows it. + getSendBuffer().appendMaskedBinaryFrame(dataPtr, length); sentFrames.incrementAndGet(); receivesSinceSend = 0; } @@ -476,6 +512,8 @@ public void batchAppendBinary(long dataPtr, int length) { @Override public void batchFlush() { writes.incrementAndGet(); + maxBufferCapacity = Math.max(maxBufferCapacity, getSendBuffer().getCapacity()); + getSendBuffer().reset(); } // The one-frame-per-write path: one frame, one write. From 8506ed591a94bf9c328434669dc84aa2af84f25a Mon Sep 17 00:00:00 2001 From: Jaromir Hamala Date: Wed, 4 Nov 2026 22:44:36 +0100 Subject: [PATCH 05/13] perf(qwp): skip the auto-flush age check on the first pending row sendRow() had just timestamped that row, so its age is zero and the check could never fire; it cost a clock read per flush for per-row producers (~10% of producer throughput at saturation). --- .../qwp/client/QwpWebSocketSender.java | 18 ++- .../client/AutoFlushIntervalClockTest.java | 105 ++++++++++++++++++ 2 files changed, 120 insertions(+), 3 deletions(-) create mode 100644 core/src/test/java/io/questdb/client/test/cutlass/qwp/client/AutoFlushIntervalClockTest.java diff --git a/core/src/main/java/io/questdb/client/cutlass/qwp/client/QwpWebSocketSender.java b/core/src/main/java/io/questdb/client/cutlass/qwp/client/QwpWebSocketSender.java index 6b8eec79..707164be 100644 --- a/core/src/main/java/io/questdb/client/cutlass/qwp/client/QwpWebSocketSender.java +++ b/core/src/main/java/io/questdb/client/cutlass/qwp/client/QwpWebSocketSender.java @@ -86,6 +86,7 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicReference; import java.util.concurrent.locks.ReentrantLock; +import java.util.function.LongSupplier; import java.util.function.Supplier; /** @@ -374,6 +375,9 @@ public class QwpWebSocketSender implements Sender { private SenderErrorHandler errorHandler = DefaultSenderErrorHandler.INSTANCE; private int errorInboxCapacity = SenderErrorDispatcher.DEFAULT_CAPACITY; private long firstPendingRowTimeNanos; + // Clock for the auto-flush age check only (sendRow / shouldAutoFlush). + // Test seam: setNanoClockForTesting swaps it to count or steer the reads. + private LongSupplier nanoClock = System::nanoTime; // Additive offset applied to every user-visible FSN this sender reports // (flushAndGetSequence, awaitAckedFsn's target, getAckedFsn, drain's // watermark, and every FSN the I/O loop surfaces through the progress @@ -2527,6 +2531,11 @@ public void setChunkPublishFaultForTesting(Runnable fault) { this.chunkPublishFaultForTesting = fault; } + @TestOnly + public void setNanoClockForTesting(LongSupplier nanoClock) { + this.nanoClock = nanoClock; + } + /** * Background orphan-drainer pool, or {@code null} when * {@code drain_orphans} is off or no orphan slot was adopted. @@ -7133,7 +7142,7 @@ private void sendRow() { currentTableBufferSnapshotBytes, cap > 0 ? cap : Long.MAX_VALUE); if (pendingRowCount == 0) { - firstPendingRowTimeNanos = System.nanoTime(); + firstPendingRowTimeNanos = nanoClock.getAsLong(); } pendingRowCount++; @@ -7164,8 +7173,11 @@ private boolean shouldAutoFlush() { if (effectiveAutoFlushBytes > 0 && getPendingBytes() >= effectiveAutoFlushBytes) { return true; } - if (autoFlushIntervalNanos > 0) { - long ageNanos = System.nanoTime() - firstPendingRowTimeNanos; + // pendingRowCount == 1: sendRow() just timestamped this very row, so its + // age is ~0 and can never reach the (>= 1 ms) interval. Skipping the + // clock read there saves one nanoTime() per flush for per-row flushers. + if (autoFlushIntervalNanos > 0 && pendingRowCount > 1) { + long ageNanos = nanoClock.getAsLong() - firstPendingRowTimeNanos; return ageNanos >= autoFlushIntervalNanos; } return false; diff --git a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/AutoFlushIntervalClockTest.java b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/AutoFlushIntervalClockTest.java new file mode 100644 index 00000000..445fbdc4 --- /dev/null +++ b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/AutoFlushIntervalClockTest.java @@ -0,0 +1,105 @@ +/*+***************************************************************************** + * ___ _ ____ ____ + * / _ \ _ _ ___ ___| |_| _ \| __ ) + * | | | | | | |/ _ \/ __| __| | | | _ \ + * | |_| | |_| | __/\__ \ |_| |_| | |_) | + * \__\_\\__,_|\___||___/\__|____/|____/ + * + * Copyright (c) 2014-2019 Appsicle + * Copyright (c) 2019-2026 QuestDB + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + ******************************************************************************/ + +package io.questdb.client.test.cutlass.qwp.client; + +import io.questdb.client.Sender; +import io.questdb.client.cutlass.qwp.client.QwpWebSocketSender; +import io.questdb.client.test.cutlass.qwp.websocket.TestWebSocketServer; +import org.junit.Assert; +import org.junit.Test; + +import java.io.IOException; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; + +import static io.questdb.client.test.tools.TestUtils.assertMemoryLeak; + +/** + * Pins how often the WebSocket sender reads its clock on the row path, via the + * injectable auto-flush clock, and that interval-based auto-flush still fires. + */ +public class AutoFlushIntervalClockTest { + private static final int N = 1000; + + @Test + public void testIntervalFlushesOnSecondRowOfAgedBatch() throws Exception { + assertMemoryLeak(() -> { + try (TestWebSocketServer server = startAckingServer(); + Sender sender = Sender.fromConfig(cfg(server) + "auto_flush_rows=1000000;auto_flush_interval=100;")) { + QwpWebSocketSender ws = (QwpWebSocketSender) sender; + AtomicLong now = new AtomicLong(); + ws.setNanoClockForTesting(now::get); + long fsn0 = ws.getCursorEngineForTesting().publishedFsn(); + + sender.table("t").longColumn("v", 1L).atNow(); + Assert.assertEquals("first row must stay buffered", + fsn0, ws.getCursorEngineForTesting().publishedFsn()); + + now.addAndGet(TimeUnit.MILLISECONDS.toNanos(200)); + sender.table("t").longColumn("v", 2L).atNow(); + Assert.assertTrue("batch older than the interval must auto-flush on its second row", + ws.getCursorEngineForTesting().publishedFsn() > fsn0); + } + }); + } + + @Test + public void testOneClockReadPerSingleRowFlush() throws Exception { + assertMemoryLeak(() -> { + try (TestWebSocketServer server = startAckingServer(); + Sender sender = Sender.fromConfig(cfg(server) + "auto_flush_rows=1000000;auto_flush_interval=100;")) { + AtomicLong reads = new AtomicLong(); + ((QwpWebSocketSender) sender).setNanoClockForTesting(() -> reads.incrementAndGet() * 1_000L); + for (int i = 0; i < N; i++) { + sender.table("t").longColumn("v", i).atNow(); + sender.flush(); + } + Assert.assertEquals("one clock read (the batch timestamp) per single-row flush", + N, reads.get()); + } + }); + } + + private static String cfg(TestWebSocketServer server) { + return "ws::addr=localhost:" + server.getPort() + ";"; + } + + private static TestWebSocketServer startAckingServer() throws Exception { + AtomicLong seq = new AtomicLong(); + TestWebSocketServer server = new TestWebSocketServer(new TestWebSocketServer.WebSocketServerHandler() { + @Override + public void onBinaryMessage(TestWebSocketServer.ClientHandler client, byte[] data) { + try { + client.sendBinary(CloseDrainTest.buildAck(seq.getAndIncrement())); + } catch (IOException e) { + throw new RuntimeException(e); + } + } + }); + server.start(); + Assert.assertTrue(server.awaitStart(5, TimeUnit.SECONDS)); + return server; + } +} From e4f3338b1afc896d5c9fffc2aea16ea0c498e712 Mon Sep 17 00:00:00 2001 From: Jaromir Hamala Date: Wed, 4 Nov 2026 22:48:04 +0100 Subject: [PATCH 06/13] feat(qwp): allow auto_flush_interval=off for the WebSocket transport Explicit opt-out only; the default interval is unchanged. With the interval off, row-cap and byte-cap auto-flush and explicit flush() still apply, and the sender reads no clock on the row path at all (measured +21% producer throughput for per-row flushers together with the previous commit). --- .../main/java/io/questdb/client/Sender.java | 14 +++-- .../qwp/client/QwpWebSocketSender.java | 5 +- .../client/test/QuestDBBuilderTest.java | 18 +++--- .../client/AutoFlushIntervalClockTest.java | 60 +++++++++++++++++++ .../qwp/client/DeltaDictRecoveryTest.java | 2 - .../LineSenderBuilderWebSocketTest.java | 21 +++++-- .../SymbolDictRecycleFsnContinuityTest.java | 6 +- .../test/impl/WsSenderConfigHonoredTest.java | 3 +- 8 files changed, 100 insertions(+), 29 deletions(-) diff --git a/core/src/main/java/io/questdb/client/Sender.java b/core/src/main/java/io/questdb/client/Sender.java index d0dab97c..9c9c31e9 100644 --- a/core/src/main/java/io/questdb/client/Sender.java +++ b/core/src/main/java/io/questdb/client/Sender.java @@ -3561,9 +3561,14 @@ private QwpWebSocketSender buildWebSocket(HttpTokenProvider tokenProvider) { int actualAutoFlushRows = autoFlushRows == PARAMETER_NOT_SET_EXPLICITLY ? DEFAULT_WS_AUTO_FLUSH_ROWS : autoFlushRows; int actualAutoFlushBytes = autoFlushBytes == PARAMETER_NOT_SET_EXPLICITLY ? DEFAULT_WS_AUTO_FLUSH_BYTES : autoFlushBytes; - long actualAutoFlushIntervalNanos = autoFlushIntervalMillis == PARAMETER_NOT_SET_EXPLICITLY - ? DEFAULT_WS_AUTO_FLUSH_INTERVAL_NANOS - : TimeUnit.MILLISECONDS.toNanos(autoFlushIntervalMillis); + long actualAutoFlushIntervalNanos; + if (autoFlushIntervalMillis == PARAMETER_NOT_SET_EXPLICITLY) { + actualAutoFlushIntervalNanos = DEFAULT_WS_AUTO_FLUSH_INTERVAL_NANOS; + } else if (autoFlushIntervalMillis == Integer.MAX_VALUE) { + actualAutoFlushIntervalNanos = 0; // explicit off: QwpWebSocketSender treats 0 as disabled + } else { + actualAutoFlushIntervalNanos = TimeUnit.MILLISECONDS.toNanos(autoFlushIntervalMillis); + } Supplier wsAuthHeader = buildWebSocketAuthHeader(tokenProvider); @@ -5057,9 +5062,6 @@ private void validateParameters() { if (protocolVersion != PARAMETER_NOT_SET_EXPLICITLY) { throw new LineSenderException("protocol version is not supported for WebSocket protocol"); } - if (autoFlushIntervalMillis == Integer.MAX_VALUE) { - throw new LineSenderException("disabling auto-flush is not supported for WebSocket protocol"); - } if ((sfDurability == SfDurability.FLUSH || sfDurability == SfDurability.APPEND)) { throw new LineSenderException( "sf_durability=" + sfDurability.name().toLowerCase() diff --git a/core/src/main/java/io/questdb/client/cutlass/qwp/client/QwpWebSocketSender.java b/core/src/main/java/io/questdb/client/cutlass/qwp/client/QwpWebSocketSender.java index 707164be..229d73ba 100644 --- a/core/src/main/java/io/questdb/client/cutlass/qwp/client/QwpWebSocketSender.java +++ b/core/src/main/java/io/questdb/client/cutlass/qwp/client/QwpWebSocketSender.java @@ -2457,7 +2457,7 @@ public int getAutoFlushBytes() { } /** - * Returns the auto-flush interval in nanoseconds. + * Returns the auto-flush interval in nanoseconds (0 = interval-based auto-flush disabled). */ public long getAutoFlushIntervalNanos() { return autoFlushIntervalNanos; @@ -7141,7 +7141,8 @@ private void sendRow() { long bufferedNow = currentTableBuffer.nextRow( currentTableBufferSnapshotBytes, cap > 0 ? cap : Long.MAX_VALUE); - if (pendingRowCount == 0) { + // Interval off (0): no clock read on the row path at all. + if (pendingRowCount == 0 && autoFlushIntervalNanos > 0) { firstPendingRowTimeNanos = nanoClock.getAsLong(); } pendingRowCount++; diff --git a/core/src/test/java/io/questdb/client/test/QuestDBBuilderTest.java b/core/src/test/java/io/questdb/client/test/QuestDBBuilderTest.java index e338aa72..97a480b1 100644 --- a/core/src/test/java/io/questdb/client/test/QuestDBBuilderTest.java +++ b/core/src/test/java/io/questdb/client/test/QuestDBBuilderTest.java @@ -212,23 +212,27 @@ public void testMalformedEgressConfigRejectedAtBuildWithMinZero() { "compression_level"); } + @Test + public void testAutoFlushOffAcceptedAtBuildWithMinZero() { + // auto_flush_interval=off and auto_flush=off are legal on WebSocket. + QuestDB.builder().fromConfig( + "ws::addr=127.0.0.1:1;auto_flush_interval=off;sender_pool_min=0;query_pool_min=0;").build().close(); + QuestDB.builder().fromConfig( + "ws::addr=127.0.0.1:1;auto_flush=off;sender_pool_min=0;query_pool_min=0;").build().close(); + } + @Test public void testMalformedIngressConfigRejectedAtBuildWithMinZero() { // sender_pool_min=0 pre-warms nothing, so build() never constructs a // Sender -- yet it must still reject a malformed ingress key in the // single cluster config up front. Covers a typed enum (tls_verify), a // registry-STRING value that only the real Sender parse validates - // (auto_flush_rows), and WebSocket build-time checks: auto_flush=off and - // auto_flush_interval=off both disable auto-flush (unsupported on - // WebSocket), and sf_durability=flush is not yet supported. + // (auto_flush_rows), and a WebSocket build-time check: sf_durability=flush + // is not yet supported. assertBuildRejected( "wss::addr=127.0.0.1:1;tls_verify=strict;sender_pool_min=0;query_pool_min=0;", "tls_verify"); assertBuildRejected( "ws::addr=127.0.0.1:1;auto_flush_rows=abc;sender_pool_min=0;query_pool_min=0;", "auto_flush_rows"); - assertBuildRejected( - "ws::addr=127.0.0.1:1;auto_flush_interval=off;sender_pool_min=0;query_pool_min=0;", "auto-flush"); - assertBuildRejected( - "ws::addr=127.0.0.1:1;auto_flush=off;sender_pool_min=0;query_pool_min=0;", "auto-flush"); assertBuildRejected( "ws::addr=127.0.0.1:1;sf_durability=flush;sender_pool_min=0;query_pool_min=0;", "not yet supported"); } diff --git a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/AutoFlushIntervalClockTest.java b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/AutoFlushIntervalClockTest.java index 445fbdc4..b2522295 100644 --- a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/AutoFlushIntervalClockTest.java +++ b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/AutoFlushIntervalClockTest.java @@ -43,6 +43,22 @@ public class AutoFlushIntervalClockTest { private static final int N = 1000; + @Test + public void testDefaultIntervalWhenKeyAbsentAndZeroWhenOff() throws Exception { + assertMemoryLeak(() -> { + try (TestWebSocketServer server = startAckingServer()) { + try (Sender sender = Sender.fromConfig(cfg(server))) { + Assert.assertEquals("ws:: without auto_flush_interval keeps the 100 ms default", + TimeUnit.MILLISECONDS.toNanos(100), + ((QwpWebSocketSender) sender).getAutoFlushIntervalNanos()); + } + try (Sender sender = Sender.fromConfig(cfg(server) + "auto_flush_interval=off;")) { + Assert.assertEquals(0L, ((QwpWebSocketSender) sender).getAutoFlushIntervalNanos()); + } + } + }); + } + @Test public void testIntervalFlushesOnSecondRowOfAgedBatch() throws Exception { assertMemoryLeak(() -> { @@ -65,6 +81,50 @@ public void testIntervalFlushesOnSecondRowOfAgedBatch() throws Exception { }); } + @Test + public void testIntervalOffReadsNoClock() throws Exception { + assertMemoryLeak(() -> { + try (TestWebSocketServer server = startAckingServer(); + Sender sender = Sender.fromConfig(cfg(server) + "auto_flush_interval=off;")) { + AtomicLong reads = new AtomicLong(); + ((QwpWebSocketSender) sender).setNanoClockForTesting(() -> reads.incrementAndGet() * 1_000L); + for (int i = 0; i < N; i++) { + sender.table("t").longColumn("v", i).atNow(); + sender.flush(); + } + Assert.assertEquals("auto_flush_interval=off must not read the clock on the row path", + 0L, reads.get()); + } + }); + } + + @Test + public void testIntervalOffKeepsRowCapAndExplicitFlush() throws Exception { + assertMemoryLeak(() -> { + try (TestWebSocketServer server = startAckingServer(); + Sender sender = Sender.fromConfig(cfg(server) + "auto_flush_interval=off;auto_flush_rows=10;")) { + QwpWebSocketSender ws = (QwpWebSocketSender) sender; + long fsn0 = ws.getCursorEngineForTesting().publishedFsn(); + for (int i = 0; i < 9; i++) { + sender.table("t").longColumn("v", i).atNow(); + } + Assert.assertEquals("9 rows stay below the row cap", + fsn0, ws.getCursorEngineForTesting().publishedFsn()); + sender.table("t").longColumn("v", 9L).atNow(); + long fsn1 = ws.getCursorEngineForTesting().publishedFsn(); + Assert.assertTrue("the 10th row must trigger the row-cap flush", fsn1 > fsn0); + + for (int i = 0; i < 3; i++) { + sender.table("t").longColumn("v", i).atNow(); + } + Assert.assertEquals(fsn1, ws.getCursorEngineForTesting().publishedFsn()); + sender.flush(); + Assert.assertTrue("explicit flush() must send the partial batch", + ws.getCursorEngineForTesting().publishedFsn() > fsn1); + } + }); + } + @Test public void testOneClockReadPerSingleRowFlush() throws Exception { assertMemoryLeak(() -> { diff --git a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/DeltaDictRecoveryTest.java b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/DeltaDictRecoveryTest.java index fbc833d8..3b8d316c 100644 --- a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/DeltaDictRecoveryTest.java +++ b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/DeltaDictRecoveryTest.java @@ -1118,8 +1118,6 @@ public void testFailedPublishThenNewSymbolPersistsSuffixWithoutDuplicating() thr // atNow() and throws from there instead of from the flush() this test // brackets. The batch and the branch under test are identical either way; // only the throw site moves, so parking the thresholds costs no coverage. - // (WebSocket rejects auto_flush=off outright, hence the large finite - // interval.) String cfg = "ws::addr=localhost:" + port + ";sf_dir=" + sfDir + ";sf_max_segment_bytes=1024" + ";auto_flush_bytes=off;auto_flush_rows=1000000;auto_flush_interval=60000;"; diff --git a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/LineSenderBuilderWebSocketTest.java b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/LineSenderBuilderWebSocketTest.java index 1d83dff2..c5dad492 100644 --- a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/LineSenderBuilderWebSocketTest.java +++ b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/LineSenderBuilderWebSocketTest.java @@ -425,12 +425,21 @@ public void testDefaultIsAsync() { } @Test - public void testDisableAutoFlush_notSupportedForWebSocket() { - assertThrowsAny( - Sender.builder(Sender.Transport.WEBSOCKET) - .address(LOCALHOST) - .disableAutoFlush(), - "not supported for WebSocket"); + public void testDisableAutoFlush_supportedForWebSocket() throws Exception { + assertMemoryLeak(() -> { + try (TestWebSocketServer server = new TestWebSocketServer(new TestWebSocketServer.WebSocketServerHandler() { + })) { + server.start(); + Assert.assertTrue(server.awaitStart(5, TimeUnit.SECONDS)); + try (QwpWebSocketSender sender = (QwpWebSocketSender) Sender.builder(Sender.Transport.WEBSOCKET) + .address(LOCALHOST + ":" + server.getPort()) + .disableAutoFlush() + .build()) { + Assert.assertEquals(0, sender.getAutoFlushRows()); + Assert.assertEquals(0L, sender.getAutoFlushIntervalNanos()); + } + } + }); } @Test diff --git a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/SymbolDictRecycleFsnContinuityTest.java b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/SymbolDictRecycleFsnContinuityTest.java index 69520d31..93032ae6 100644 --- a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/SymbolDictRecycleFsnContinuityTest.java +++ b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/SymbolDictRecycleFsnContinuityTest.java @@ -392,10 +392,8 @@ public void testLatchedErrorStillThrowsForOldEpochTarget() throws Exception { * "greater than" check but fails the exact equalities below. Arming uses * resetSymbolDictionary() so the test does not depend on threshold or * anti-thrash floor arithmetic; auto-flush stays off so no interval - * flush can inject frames on a slow CI box. WebSocket rejects - * {@code auto_flush=off} outright (it also zeroes the auto-flush - * interval, which {@code validateParameters} disallows for this - * transport), so this pins the same large-finite-interval idiom already + * flush can inject frames on a slow CI box. This pins the same + * large-finite-interval idiom already * used by {@code DeltaDictRecoveryTest} / {@code SelfSufficientFramesTest} * to fully suppress auto-flush on WS. */ diff --git a/core/src/test/java/io/questdb/client/test/impl/WsSenderConfigHonoredTest.java b/core/src/test/java/io/questdb/client/test/impl/WsSenderConfigHonoredTest.java index 39695ec2..e4b94d8f 100644 --- a/core/src/test/java/io/questdb/client/test/impl/WsSenderConfigHonoredTest.java +++ b/core/src/test/java/io/questdb/client/test/impl/WsSenderConfigHonoredTest.java @@ -54,8 +54,7 @@ public void testEveryIngressKeyIsHonored() { assertHonored("auto_flush_interval=250", "auto_flush_interval", 250); // auto_flush=off reaches disableAutoFlush(), which sets the interval to // MAX_VALUE; auto_flush_rows=off leaves the interval unset, so this field - // distinguishes the two. That config is build-rejected on WebSocket; see - // QuestDBBuilderTest.testMalformedIngressConfigRejectedAtBuildWithMinZero. + // distinguishes the two. assertHonored("auto_flush=off", "auto_flush_interval", Integer.MAX_VALUE); assertHonored("max_name_len=99", "max_name_len", 99); assertHonored("transaction=on", "transaction", true); From 05b31067cb1dfc1297220442a60e74a7527ee850 Mon Sep 17 00:00:00 2001 From: Jaromir Hamala Date: Wed, 4 Nov 2026 22:55:50 +0100 Subject: [PATCH 07/13] docs(qwp): disableAutoFlush() javadoc covers the WebSocket transport --- core/src/main/java/io/questdb/client/Sender.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/core/src/main/java/io/questdb/client/Sender.java b/core/src/main/java/io/questdb/client/Sender.java index 9c9c31e9..9cbf5b99 100644 --- a/core/src/main/java/io/questdb/client/Sender.java +++ b/core/src/main/java/io/questdb/client/Sender.java @@ -1864,8 +1864,8 @@ public LineSenderBuilder connectionListenerInboxCapacity(int capacity) { *

* It is essential to ensure the maximum buffer capacity is sufficient to accommodate all locally buffered data. *

- * This method should only be used when communicating via the HTTP protocol. Calling this method is illegal when - * communicating over the TCP protocol. + * This method is supported for the HTTP and WebSocket (QWP) protocols. Calling this method is illegal when + * communicating over the TCP protocol. On WebSocket, {@link Sender#close()} still flushes pending rows. * * @return this instance for method chaining * @see #autoFlushRows(int) From 9f7defdc1262f5c381199592ecbaf84f3c339fed Mon Sep 17 00:00:00 2001 From: Jaromir Hamala Date: Thu, 5 Nov 2026 13:00:54 +0100 Subject: [PATCH 08/13] fix(qwp): exclude failed send batches from replay metrics --- .../qwp/client/QwpWebSocketSender.java | 8 +- .../sf/cursor/CursorWebSocketSendLoop.java | 22 +- ...sorWebSocketSendLoopReplayMetricsTest.java | 251 ++++++++++++++++++ 3 files changed, 273 insertions(+), 8 deletions(-) create mode 100644 core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoopReplayMetricsTest.java diff --git a/core/src/main/java/io/questdb/client/cutlass/qwp/client/QwpWebSocketSender.java b/core/src/main/java/io/questdb/client/cutlass/qwp/client/QwpWebSocketSender.java index 229d73ba..4c8e731a 100644 --- a/core/src/main/java/io/questdb/client/cutlass/qwp/client/QwpWebSocketSender.java +++ b/core/src/main/java/io/questdb/client/cutlass/qwp/client/QwpWebSocketSender.java @@ -2917,6 +2917,8 @@ public long getTotalErrorNotificationsDelivered() { * Count of frames re-sent during post-reconnect catch-up windows since * this sender started. Zero in steady state; a sustained nonzero rate * signals flapping where every reconnect replays meaningful work. + * A frame counts as replayed only if an earlier send completed successfully. + * Failed batch flushes do not count, even if some bytes were written. * Sender-lifetime: carried across symbol-dictionary recycles and retained * after {@link #close()}. */ @@ -2925,8 +2927,10 @@ public long getTotalFramesReplayed() { } /** - * Binary frames the cursor I/O loop has issued to the wire since this - * sender started, replays included. Sender-lifetime: carried across + * Binary frames the cursor I/O loop has successfully sent since this + * sender started, replays included. A batch contributes only when its + * entire flush succeeds; failed flushes contribute no frames, even if + * some bytes were written. Sender-lifetime: carried across * symbol-dictionary recycles and retained after {@link #close()}. */ public long getTotalFramesSent() { 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 index 47199487..94ab8f40 100644 --- 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 @@ -1682,7 +1682,8 @@ public long getTotalDurableTrimAdvances() { * Cumulative count of frames re-sent during post-reconnect catch-up * windows. One increment per replayed frame. Zero in steady state; a * sustained nonzero rate signals flapping where every reconnect replays - * meaningful work. + * meaningful work. Failed batch flushes neither increase this count nor + * establish a replay boundary, even if some bytes were written. */ public long getTotalFramesReplayed() { return counters.framesReplayed.get(); @@ -3109,11 +3110,12 @@ private void sendPingIfDue(long intervalNanos) { /** * Freezes the highest FSN that completed a send on the connection which - * just failed. Frames published before the first connection, or while a - * reconnect is in progress, have not been sent and therefore do not belong - * to the replay metric. Keep an older, higher target when another failure - * interrupts catch-up so the remaining frames still count as re-sends on - * the following connection. + * just failed. A batch counts only when its entire flush succeeds, even + * if a failed flush wrote some bytes. Frames published before the first + * connection, or while a reconnect is in progress, have not been sent and + * therefore do not belong to the replay metric. Keep an older, higher + * target when another failure interrupts catch-up so the remaining frames + * still count as re-sends on the following connection. */ private void snapshotReplayTarget() { if (hasEverConnected && nextWireSeq > 0L) { @@ -3891,6 +3893,10 @@ private boolean tryReceiveAcks() { * one-frame-per-call path did after its previous frame went out. */ private boolean trySendBatch() { + // Appending advances these speculatively; reconnect must snapshot only + // completed flushes and retain any older replay target after a failure. + final long batchStartWireSeq = nextWireSeq; + final long batchStartReplayTargetFsn = replayTargetFsn; int batchFrames = 0; int batchReplayed = 0; long batchBytes = 0; @@ -4028,6 +4034,8 @@ private boolean trySendBatch() { try { client.batchAppendBinary(frameAddr, payloadLen); } catch (Throwable t) { + nextWireSeq = batchStartWireSeq; + replayTargetFsn = batchStartReplayTargetFsn; fail(t); return false; } @@ -4072,6 +4080,8 @@ private boolean trySendBatch() { try { client.batchFlush(); } catch (Throwable t) { + nextWireSeq = batchStartWireSeq; + replayTargetFsn = batchStartReplayTargetFsn; fail(t); return false; } diff --git a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoopReplayMetricsTest.java b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoopReplayMetricsTest.java new file mode 100644 index 00000000..4c13f61a --- /dev/null +++ b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoopReplayMetricsTest.java @@ -0,0 +1,251 @@ +/*+***************************************************************************** + * ___ _ ____ ____ + * / _ \ _ _ ___ ___| |_| _ \| __ ) + * | | | | | | |/ _ \/ __| __| | | | _ \ + * | |_| | |_| | __/\__ \ |_| |_| | |_) | + * \__\_\\__,_|\___||___/\__|____/|____/ + * + * Copyright (c) 2014-2019 Appsicle + * Copyright (c) 2019-2026 QuestDB + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + ******************************************************************************/ + +package io.questdb.client.test.cutlass.qwp.client.sf.cursor; + +import io.questdb.client.DefaultHttpClientConfiguration; +import io.questdb.client.cutlass.http.client.HttpClientException; +import io.questdb.client.cutlass.qwp.client.sf.cursor.CursorSendEngine; +import io.questdb.client.cutlass.qwp.client.sf.cursor.CursorWebSocketSendLoop; +import io.questdb.client.cutlass.qwp.websocket.WebSocketFrameWriter; +import io.questdb.client.network.Socket; +import io.questdb.client.network.SocketReadinessWaiter; +import io.questdb.client.std.MemoryTag; +import io.questdb.client.std.Unsafe; +import io.questdb.client.test.tools.TestUtils; +import io.questdb.client.test.tools.TestWebSocketClient; +import org.junit.Assert; +import org.junit.Test; + +public class CursorWebSocketSendLoopReplayMetricsTest { + private static final int FRAME_COUNT = 4; + private static final int PAYLOAD_LEN = 8; + private static final int WIRE_LEN = PAYLOAD_LEN + WebSocketFrameWriter.headerSize(PAYLOAD_LEN, true); + + @Test + public void testAppendFailureDoesNotCountBufferedFramesAsReplays() throws Exception { + assertFailedBatchDoesNotCountAsSent(0, 3); + } + + @Test + public void testPartialFlushDoesNotCountFailedBatchAsSent() throws Exception { + // Even one complete frame accepted by the socket does not complete the batch. + assertFailedBatchDoesNotCountAsSent(WIRE_LEN, Integer.MAX_VALUE); + } + + @Test + public void testReplayTargetSurvivesAnotherFailedFlush() throws Exception { + TestUtils.assertMemoryLeak(() -> { + ScriptedSocket originalSocket = new ScriptedSocket(Integer.MAX_VALUE); + ScriptedSocket retrySocket = new ScriptedSocket(2 * WIRE_LEN); + ScriptedSocket replacementSocket = new ScriptedSocket(Integer.MAX_VALUE); + try (CursorSendEngine engine = new CursorSendEngine(null, 16_384); + BatchingClient original = new BatchingClient(originalSocket); + BatchingClient retry = new BatchingClient(retrySocket); + BatchingClient replacement = new BatchingClient(replacementSocket)) { + appendFrames(engine); + retry.batchCap = 2 * WIRE_LEN; + BatchingClient[] reconnectClients = {retry, replacement}; + int[] reconnectIndex = {0}; + try (CursorWebSocketSendLoop loop = new CursorWebSocketSendLoop( + original, engine, 0L, () -> reconnectClients[reconnectIndex[0]++], 0L, 0L)) { + loop.setRunningForTest(true); + loop.positionCursorForStartForTest(); + Assert.assertTrue(loop.trySendBatchForTest()); + Assert.assertEquals(FRAME_COUNT, loop.getTotalFramesSent()); + Assert.assertEquals(0, loop.getTotalFramesReplayed()); + + loop.connectLoopForTest(new HttpClientException("test disconnect"), "reconnect", 0L); + Assert.assertTrue(loop.trySendBatchForTest()); + Assert.assertEquals(2, loop.getTotalFramesReplayed()); + // Appending the remaining two frames clears replayTargetFsn before + // this flush fails. The next connection must still replay all four. + Assert.assertFalse(loop.trySendBatchForTest()); + Assert.assertEquals(2 * WIRE_LEN, retrySocket.acceptedBytes); + Assert.assertEquals(FRAME_COUNT + 2, loop.getTotalFramesSent()); + Assert.assertEquals(2, loop.getTotalFramesReplayed()); + + Assert.assertTrue(loop.trySendBatchForTest()); + Assert.assertEquals(FRAME_COUNT * WIRE_LEN, replacementSocket.acceptedBytes); + Assert.assertEquals(2 * FRAME_COUNT + 2, loop.getTotalFramesSent()); + Assert.assertEquals(FRAME_COUNT + 2, loop.getTotalFramesReplayed()); + Assert.assertEquals(2, loop.getTotalReconnects()); + loop.checkError(); + } + } + }); + } + + @Test + public void testZeroByteFlushFailureDoesNotCountFirstDeliveryAsReplay() throws Exception { + assertFailedBatchDoesNotCountAsSent(0, Integer.MAX_VALUE); + } + + private static void appendFrames(CursorSendEngine engine) { + long payload = Unsafe.malloc(PAYLOAD_LEN, MemoryTag.NATIVE_DEFAULT); + try { + for (int i = 0; i < FRAME_COUNT; i++) { + Unsafe.getUnsafe().putLong(payload, i); + Assert.assertEquals(i, engine.appendBlocking(payload, PAYLOAD_LEN)); + } + } finally { + Unsafe.free(payload, PAYLOAD_LEN, MemoryTag.NATIVE_DEFAULT); + } + } + + private static void assertFailedBatchDoesNotCountAsSent(int acceptedBytes, int failAppendAt) throws Exception { + TestUtils.assertMemoryLeak(() -> { + ScriptedSocket firstSocket = new ScriptedSocket(acceptedBytes); + ScriptedSocket replacementSocket = new ScriptedSocket(Integer.MAX_VALUE); + try (CursorSendEngine engine = new CursorSendEngine(null, 16_384); + BatchingClient first = new BatchingClient(firstSocket); + BatchingClient replacement = new BatchingClient(replacementSocket)) { + appendFrames(engine); + first.failAppendAt = failAppendAt; + try (CursorWebSocketSendLoop loop = new CursorWebSocketSendLoop( + first, engine, 0L, () -> replacement, 0L, 0L)) { + loop.setRunningForTest(true); + loop.positionCursorForStartForTest(); + Assert.assertFalse(loop.trySendBatchForTest()); + Assert.assertEquals(acceptedBytes, firstSocket.acceptedBytes); + Assert.assertEquals(Math.min(FRAME_COUNT, failAppendAt), first.appendCalls); + if (failAppendAt <= FRAME_COUNT) { + Assert.assertEquals("append failure must happen before any socket write", 0, firstSocket.sendCalls); + } else { + Assert.assertTrue("flush must reach the socket", firstSocket.sendCalls > 0); + } + Assert.assertEquals(1, loop.getTotalReconnects()); + Assert.assertEquals(0, loop.getTotalFramesSent()); + Assert.assertEquals(0, loop.getTotalFramesReplayed()); + + Assert.assertTrue(loop.trySendBatchForTest()); + Assert.assertEquals(FRAME_COUNT * WIRE_LEN, replacementSocket.acceptedBytes); + Assert.assertEquals(FRAME_COUNT, loop.getTotalFramesSent()); + Assert.assertEquals("failed batches must not establish a replay boundary", + 0, loop.getTotalFramesReplayed()); + Assert.assertFalse("all queued frames must have been sent", loop.trySendBatchForTest()); + loop.checkError(); + } + } + }); + } + + private static class BatchingClient extends TestWebSocketClient { + private int appendCalls; + private int batchCap = super.batchByteCap(); + private int failAppendAt = Integer.MAX_VALUE; + + private BatchingClient(ScriptedSocket socket) { + super(DefaultHttpClientConfiguration.INSTANCE, (nf, log) -> socket); + } + + @Override + public void batchAppendBinary(long dataPtr, int length) { + if (++appendCalls == failAppendAt) { + throw new HttpClientException("test append failure"); + } + // Skip the handshake but use the real send buffer and batchFlush/socket path. + getSendBuffer().appendMaskedBinaryFrame(dataPtr, length); + } + + @Override + public int batchByteCap() { + return batchCap; + } + + @Override + protected void ioWait(int timeout, int op) { + Assert.fail("scripted socket never blocks"); + } + + @Override + protected void setupIoWait() { + } + } + + private static class ScriptedSocket implements Socket { + private final int bytesBeforeFailure; + private int acceptedBytes; + private int sendCalls; + + private ScriptedSocket(int bytesBeforeFailure) { + this.bytesBeforeFailure = bytesBeforeFailure; + } + + @Override + public void close() { + } + + @Override + public int getFd() { + return -1; + } + + @Override + public boolean isClosed() { + return false; + } + + @Override + public void of(int fd) { + throw new UnsupportedOperationException(); + } + + @Override + public int recv(long bufferPtr, int bufferLen) { + return 0; + } + + @Override + public int send(long bufferPtr, int bufferLen) { + sendCalls++; + if (acceptedBytes == bytesBeforeFailure) { + return -1; + } + int count = Math.min(bufferLen, bytesBeforeFailure - acceptedBytes); + acceptedBytes += count; + return count; + } + + @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; + } + } +} From bd77813cf61640a7703b518da4abf99a193463a3 Mon Sep 17 00:00:00 2001 From: Jaromir Hamala Date: Thu, 5 Nov 2026 13:11:35 +0100 Subject: [PATCH 09/13] test(ws): cover blocking receive buffer compaction --- .../http/client/WebSocketClientTest.java | 127 ++++++++++-------- 1 file changed, 70 insertions(+), 57 deletions(-) diff --git a/core/src/test/java/io/questdb/client/test/cutlass/http/client/WebSocketClientTest.java b/core/src/test/java/io/questdb/client/test/cutlass/http/client/WebSocketClientTest.java index 25d0aa4e..cc8c78aa 100644 --- a/core/src/test/java/io/questdb/client/test/cutlass/http/client/WebSocketClientTest.java +++ b/core/src/test/java/io/questdb/client/test/cutlass/http/client/WebSocketClientTest.java @@ -189,64 +189,14 @@ public void testExtractMaxBatchSizeParsesPositive() throws Exception { Assert.assertEquals(16 * 1024 * 1024, invokeExtractMaxBatchSize(response)); } - /** - * One recv delivering a burst of small frames (the shape of a server's - * cumulative ACKs) must be parsed in place: compacting the unread tail - * after every frame memmoved the rest of the burst once per frame. The - * burst fills the 64 KiB buffer and ends mid-frame, so the next recv must - * compact that tail BEFORE the grow check -- a tail left at the buffer end - * would read as "full" and grow the buffer for no reason. - */ @Test public void testFrameBurstIsParsedInPlace() throws Exception { - assertMemoryLeak(() -> { - final int frameLen = 13; // FIN|BINARY, len=11 - final int frameCount = 5100; // > 64 KiB, so the first recv ends mid-frame - byte[] data = new byte[frameLen * frameCount]; - for (int i = 0; i < frameCount; i++) { - data[i * frameLen] = (byte) 0x82; - data[i * frameLen + 1] = 11; - data[i * frameLen + 2] = (byte) i; - data[i * frameLen + 3] = (byte) (i >> 8); - } - try (FrameFeedWebSocketClient client = new FrameFeedWebSocketClient(data)) { - setUpgradedTrue(client); - final int bufSize = getIntField(client, "recvBufSize"); - final long[] payload = {0, 0}; // first, last - final int[] frames = {0}; - WebSocketFrameHandler handler = new WebSocketFrameHandler() { - @Override - public void onBinaryMessage(long payloadPtr, int payloadLen) { - Assert.assertEquals(11, payloadLen); - int tag = (Unsafe.getUnsafe().getByte(payloadPtr) & 0xFF) - | (Unsafe.getUnsafe().getByte(payloadPtr + 1) & 0xFF) << 8; - Assert.assertEquals(frames[0], tag); - if (frames[0]++ == 0) { - payload[0] = payloadPtr; - } - payload[1] = payloadPtr; - } - - @Override - public void onClose(int code, String reason) { - Assert.fail("unexpected close frame"); - } - }; + assertFrameBurstIsParsedInPlace(false); + } - int firstRecvFrames = bufSize / frameLen; - for (int i = 0; i < firstRecvFrames; i++) { - Assert.assertTrue(client.tryReceiveFrame(handler)); - Assert.assertEquals("frame " + i + " must be parsed where recv put it", - (long) i * frameLen, payload[1] - payload[0]); - } - for (int i = firstRecvFrames; i < frameCount; i++) { - Assert.assertTrue(client.tryReceiveFrame(handler)); - } - Assert.assertEquals(frameCount, frames[0]); - Assert.assertEquals("partial tail must not grow the recv buffer", - bufSize, getIntField(client, "recvBufSize")); - } - }); + @Test + public void testFrameBurstIsParsedInPlaceWithBlockingReceive() throws Exception { + assertFrameBurstIsParsedInPlace(true); } /** @@ -544,6 +494,69 @@ public void testSendPingDoesNotClobberSendBuffer() throws Exception { }); } + /** + * One recv delivering a burst of small frames (the shape of a server's + * cumulative ACKs) must be parsed in place: compacting the unread tail + * after every frame memmoved the rest of the burst once per frame. The + * burst fills the 64 KiB buffer and ends mid-frame, so the next recv must + * compact that tail BEFORE the grow check -- a tail left at the buffer end + * would read as "full" and grow the buffer for no reason. + */ + private static void assertFrameBurstIsParsedInPlace(boolean blocking) throws Exception { + assertMemoryLeak(() -> { + final int frameLen = 13; // FIN|BINARY, len=11 + final int frameCount = 5100; // > 64 KiB, so the first recv ends mid-frame + byte[] data = new byte[frameLen * frameCount]; + for (int i = 0; i < frameCount; i++) { + data[i * frameLen] = (byte) 0x82; + data[i * frameLen + 1] = 11; + data[i * frameLen + 2] = (byte) i; + data[i * frameLen + 3] = (byte) (i >> 8); + } + try (FrameFeedWebSocketClient client = new FrameFeedWebSocketClient(data)) { + setUpgradedTrue(client); + final int bufSize = getIntField(client, "recvBufSize"); + final long[] payload = {0, 0}; // first, last + final int[] frames = {0}; + WebSocketFrameHandler handler = new WebSocketFrameHandler() { + @Override + public void onBinaryMessage(long payloadPtr, int payloadLen) { + Assert.assertEquals(11, payloadLen); + int tag = (Unsafe.getUnsafe().getByte(payloadPtr) & 0xFF) + | (Unsafe.getUnsafe().getByte(payloadPtr + 1) & 0xFF) << 8; + Assert.assertEquals(frames[0], tag); + if (frames[0]++ == 0) { + payload[0] = payloadPtr; + } + payload[1] = payloadPtr; + } + + @Override + public void onClose(int code, String reason) { + Assert.fail("unexpected close frame"); + } + }; + + int firstRecvFrames = bufSize / frameLen; + for (int i = 0; i < firstRecvFrames; i++) { + Assert.assertTrue(blocking + ? client.receiveFrame(handler, 1000) + : client.tryReceiveFrame(handler)); + Assert.assertEquals("frame " + i + " must be parsed where recv put it", + (long) i * frameLen, payload[1] - payload[0]); + } + for (int i = firstRecvFrames; i < frameCount; i++) { + Assert.assertTrue(blocking + ? client.receiveFrame(handler, 1000) + : client.tryReceiveFrame(handler)); + } + Assert.assertEquals(frameCount, frames[0]); + Assert.assertEquals("partial tail must not grow the recv buffer", + bufSize, getIntField(client, "recvBufSize")); + } + }); + } + private static void assertSendBinaryWriteWaits(CapturingSocket socket, int expectedWriteWaits) throws Exception { long payload = Unsafe.malloc(7, MemoryTag.NATIVE_DEFAULT); try (CapturingWebSocketClient client = new CapturingWebSocketClient(socket)) { @@ -773,8 +786,8 @@ 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. + * sends always succeed. Drives both receive APIs through real parsing + * and handler dispatch on the test thread. */ private static class FrameFeedWebSocketClient extends TestWebSocketClient { From d1a01c66e8ff1a61fa37b57e370a700125ac70bd Mon Sep 17 00:00:00 2001 From: Jaromir Hamala Date: Thu, 5 Nov 2026 13:17:38 +0100 Subject: [PATCH 10/13] test(qwp): cover batch flushing across recovered segments --- ...rsorWebSocketSendLoopRotationRaceTest.java | 83 +++++++++++++++++-- 1 file changed, 75 insertions(+), 8 deletions(-) diff --git a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoopRotationRaceTest.java b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoopRotationRaceTest.java index 0c0e8b57..f1fd7e0f 100644 --- a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoopRotationRaceTest.java +++ b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoopRotationRaceTest.java @@ -33,7 +33,9 @@ import io.questdb.client.cutlass.qwp.client.sf.cursor.MmapSegment; import io.questdb.client.cutlass.qwp.client.sf.cursor.SegmentBudget; import io.questdb.client.cutlass.qwp.client.sf.cursor.SegmentManager; +import io.questdb.client.cutlass.qwp.client.sf.cursor.SfManifest; import io.questdb.client.network.PlainSocketFactory; +import io.questdb.client.std.FilesFacade; import io.questdb.client.std.MemoryTag; import io.questdb.client.std.Unsafe; import io.questdb.client.test.tools.TestUtils; @@ -41,7 +43,7 @@ import org.junit.Test; /** - * Probabilistic regression test for the send-loop rotation race fixed in + * Regression tests for segment handoffs. The live-rotation test covers the race fixed in * {@code CursorWebSocketSendLoop.trySendBatch()} (PR #41): the loop could observe * a segment rotation without observing the final frame published into that * segment, and advance past it — silently dropping the frame. @@ -119,6 +121,44 @@ public void tailFrameSurvivesSegmentRotation() throws Exception { // The I/O thread is joined, so its writes to recorded[] are visible here. assertContiguous(client, n); + Assert.assertEquals(n, loop.getTotalFramesSent()); + }); + } + + @Test + public void testRecoveredSealedSegmentsFlushBeforeEmptyActive() throws Exception { + TestUtils.assertMemoryLeak(() -> { + final int sealedSegments = 3; + final int framesPerSegment = 20; + final int frames = sealedSegments * framesPerSegment; + String sfDir = TestUtils.createTmpDir("qdb-recovered-batch-"); + try { + writeRecoveredSegments(sfDir, sealedSegments, framesPerSegment); + try (CursorSendEngine engine = new CursorSendEngine(sfDir, SEGMENT_BYTES); + RecordingClient client = new RecordingClient(frames); + CursorWebSocketSendLoop loop = new CursorWebSocketSendLoop( + client, engine, 0L, null, 0L, 0L)) { + Assert.assertTrue(engine.wasRecoveredFromDisk()); + Assert.assertEquals(frames - 1, engine.publishedFsn()); + Assert.assertEquals(frames, engine.activeSegment().baseSeq()); + Assert.assertEquals(MmapSegment.HEADER_SIZE, engine.activeSegment().publishedOffset()); + loop.setRunningForTest(true); + loop.positionCursorForStartForTest(); + + // Each sealed segment needs a send pass and an advance pass. + // No producer publishes into the empty active to trigger a later flush. + for (int i = 0; i < 2 * sealedSegments + 1; i++) { + loop.trySendBatchForTest(); + loop.checkError(); + } + + assertContiguous(client, frames); + Assert.assertEquals(frames, loop.getTotalFramesSent()); + Assert.assertFalse(loop.trySendBatchForTest()); + } + } finally { + TestUtils.removeTmpDirRec(sfDir); + } }); } @@ -126,25 +166,48 @@ private static void assertContiguous(RecordingClient client, int expected) { int count = client.recordedCount; for (int i = 0; i < count; i++) { if (client.recorded[i] != i) { - Assert.fail("rotation race reordered/dropped a frame — " + Assert.fail("send loop reordered/dropped a frame — " + "fsn " + i + " on the wire was " + client.recorded[i]); } } if (count != expected) { - Assert.fail("rotation race dropped a frame — expected " + expected + Assert.fail("send loop dropped a frame — expected " + expected + " frames on the wire, got " + count); } } + private static void writeRecoveredSegments(String sfDir, int sealedSegments, int framesPerSegment) { + long payload = Unsafe.malloc(PAYLOAD_LEN, MemoryTag.NATIVE_DEFAULT); + try { + for (int s = 0; s <= sealedSegments; s++) { + String name = s == 0 ? "sf-initial.sfa" : String.format("sf-%016x.sfa", s); + long baseFsn = (long) s * framesPerSegment; + try (MmapSegment segment = MmapSegment.create(sfDir + "/" + name, baseFsn, SEGMENT_BYTES)) { + // The last segment is the empty active left by a crash after rotation. + if (s < sealedSegments) { + for (int i = 0; i < framesPerSegment; i++) { + Unsafe.getUnsafe().putLong(payload, baseFsn + i); + Assert.assertTrue(segment.tryAppend(payload, PAYLOAD_LEN) >= 0); + } + } + } + } + SfManifest.create(FilesFacade.INSTANCE, sfDir, 0, (long) sealedSegments * framesPerSegment).close(); + } finally { + Unsafe.free(payload, PAYLOAD_LEN, MemoryTag.NATIVE_DEFAULT); + } + } + /** - * In-memory transport, and nothing more: {@code batchAppendBinary} captures the fsn - * carried by each frame the loop ships. The other callbacks are inert — this + * In-memory transport: {@code batchAppendBinary} buffers each fsn and + * {@code batchFlush} publishes the delivered count. The other callbacks are inert — this * test never receives ACKs and never reconnects. Only {@code sentCount} is * read cross-thread (by the test, for progress) and so is volatile. */ private static final class RecordingClient extends TestWebSocketClient { final long[] recorded; volatile long sentCount; + private int appendedCount; private int recordedCount; RecordingClient(int capacity) { @@ -154,9 +217,13 @@ private static final class RecordingClient extends TestWebSocketClient { @Override public void batchAppendBinary(long dataPtr, int length) { - if (recordedCount < recorded.length) { - recorded[recordedCount++] = Unsafe.getUnsafe().getLong(dataPtr); - } + Assert.assertTrue("more frames than expected", appendedCount < recorded.length); + recorded[appendedCount++] = Unsafe.getUnsafe().getLong(dataPtr); + } + + @Override + public void batchFlush() { + recordedCount = appendedCount; sentCount = recordedCount; } From f643ffb150bb934c428322cfc308fc97cae80911 Mon Sep 17 00:00:00 2001 From: Jaromir Hamala Date: Thu, 5 Nov 2026 13:20:13 +0100 Subject: [PATCH 11/13] docs(qwp): correct dictionary accumulation ordering comments --- .../sf/cursor/CursorWebSocketSendLoop.java | 20 +++++++++---------- 1 file changed, 9 insertions(+), 11 deletions(-) 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 index 94ab8f40..8a76edcc 100644 --- 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 @@ -3342,10 +3342,10 @@ private void ensureSentDictCapacity(long required) { return; } if (required > MAX_SENT_DICT_BYTES) { - // Latch a terminal, do NOT just throw: accumulateSentDict runs AFTER - // the frame's sendBinary, so a bare throw unwinds to ioLoop -> fail() - // -> connectLoop, which (running still true) reconnects and replays the - // same frame, which re-overflows the never-shrinking mirror -- an + // Latch a terminal, do NOT just throw: accumulateSentDict runs after + // batchAppendBinary, before batchFlush. A bare throw unwinds to + // ioLoop -> fail() -> connectLoop, which (running still true) reconnects + // and retries the same frame, re-overflowing the never-shrinking mirror -- an // unbounded reconnect livelock rather than the "fails loudly" the // MAX_SENT_DICT_BYTES ceiling promises. recordFatal flips running=false // so connectLoop's !running guard winds the loop down and checkError() @@ -3378,13 +3378,11 @@ private void ensureSentDictCapacity(long required) { sentDictBytesOwned = true; } } catch (Throwable t) { - // Latch for exactly the reason the MAX_SENT_DICT_BYTES arm above does, and it - // was asymmetric until now: accumulateSentDict runs AFTER the frame's - // sendBinary, so a bare throw unwinds to ioLoop -> fail() -> connectLoop, - // which (running still true) reconnects and replays the same frame, which - // re-attempts the same allocation. A genuine out-of-memory therefore became an - // unbounded reconnect livelock instead of the loud failure the ceiling arm - // promises. recordFatal flips running=false so connectLoop winds down and + // Latch for the same reason as the MAX_SENT_DICT_BYTES arm above: + // accumulateSentDict runs after batchAppendBinary, before batchFlush. + // A bare throw unwinds to ioLoop -> fail() -> connectLoop, which + // (running still true) reconnects and retries the same allocation. + // recordFatal flips running=false so connectLoop winds down and // checkError() surfaces it; the throw still unwinds past the pending copy. recordFatal(t); throw t; From a2f14d28edabdca1c2c1fd478fbd999a9f605ae6 Mon Sep 17 00:00:00 2001 From: Jaromir Hamala Date: Thu, 5 Nov 2026 19:43:22 +0100 Subject: [PATCH 12/13] Keep auto_flush=off rejected on WebSocket The branch accepted auto_flush=off / disableAutoFlush() on ws:: along with auto_flush_interval=off, because both store the interval as Integer.MAX_VALUE and the WebSocket validation keyed on that value. Fully disabling auto-flush on WebSocket left auto_flush_bytes in force, so auto_flush=off;auto_flush_bytes=N still flushed every N bytes, contrary to the disableAutoFlush() contract. The builder now records disableAutoFlush() in its own flag, and the WebSocket validation rejects that flag again, as main does. Only the time-based trigger can be turned off on WebSocket, with auto_flush_interval=off; auto_flush_rows=off and auto_flush_bytes=off remain the per-trigger opt-outs. The disableAutoFlush() javadoc and the tests that the branch had flipped are restored, and QuestDBBuilderTest pins both sides: auto_flush=off is rejected, also with auto_flush_bytes, while auto_flush_interval=off is accepted alone and with the row trigger off. --- .../main/java/io/questdb/client/Sender.java | 13 ++++++++++-- .../client/test/QuestDBBuilderTest.java | 18 +++++++++++----- .../qwp/client/DeltaDictRecoveryTest.java | 2 ++ .../LineSenderBuilderWebSocketTest.java | 21 ++++++------------- .../SymbolDictRecycleFsnContinuityTest.java | 3 ++- .../test/impl/WsSenderConfigHonoredTest.java | 3 ++- 6 files changed, 36 insertions(+), 24 deletions(-) diff --git a/core/src/main/java/io/questdb/client/Sender.java b/core/src/main/java/io/questdb/client/Sender.java index 9cbf5b99..d16e9780 100644 --- a/core/src/main/java/io/questdb/client/Sender.java +++ b/core/src/main/java/io/questdb/client/Sender.java @@ -1118,6 +1118,10 @@ final class LineSenderBuilder { private final IntList ports = new IntList(); private long authTimeoutMillis = QwpWebSocketSender.DEFAULT_AUTH_TIMEOUT_MS; private int autoFlushBytes = PARAMETER_NOT_SET_EXPLICITLY; + // Set by disableAutoFlush(). auto_flush_interval=off also stores the interval as + // Integer.MAX_VALUE, so the WebSocket validation keys on this flag to reject only the + // full disable and accept the interval-only opt-out. + private boolean autoFlushDisabled; private int autoFlushIntervalMillis = PARAMETER_NOT_SET_EXPLICITLY; private int autoFlushRows = PARAMETER_NOT_SET_EXPLICITLY; private int bufferCapacity = PARAMETER_NOT_SET_EXPLICITLY; @@ -1864,8 +1868,9 @@ public LineSenderBuilder connectionListenerInboxCapacity(int capacity) { *

* It is essential to ensure the maximum buffer capacity is sufficient to accommodate all locally buffered data. *

- * This method is supported for the HTTP and WebSocket (QWP) protocols. Calling this method is illegal when - * communicating over the TCP protocol. On WebSocket, {@link Sender#close()} still flushes pending rows. + * This method should only be used when communicating via the HTTP protocol. Calling this method is illegal when + * communicating over the TCP or WebSocket protocol. On WebSocket, {@code auto_flush_interval=off} turns off + * only the time-based trigger. * * @return this instance for method chaining * @see #autoFlushRows(int) @@ -1883,6 +1888,7 @@ public LineSenderBuilder disableAutoFlush() { this.autoFlushRows = AUTO_FLUSH_DISABLED; this.autoFlushIntervalMillis = Integer.MAX_VALUE; + this.autoFlushDisabled = true; return this; } @@ -5062,6 +5068,9 @@ private void validateParameters() { if (protocolVersion != PARAMETER_NOT_SET_EXPLICITLY) { throw new LineSenderException("protocol version is not supported for WebSocket protocol"); } + if (autoFlushDisabled) { + throw new LineSenderException("disabling auto-flush is not supported for WebSocket protocol"); + } if ((sfDurability == SfDurability.FLUSH || sfDurability == SfDurability.APPEND)) { throw new LineSenderException( "sf_durability=" + sfDurability.name().toLowerCase() diff --git a/core/src/test/java/io/questdb/client/test/QuestDBBuilderTest.java b/core/src/test/java/io/questdb/client/test/QuestDBBuilderTest.java index 97a480b1..e304df95 100644 --- a/core/src/test/java/io/questdb/client/test/QuestDBBuilderTest.java +++ b/core/src/test/java/io/questdb/client/test/QuestDBBuilderTest.java @@ -213,12 +213,14 @@ public void testMalformedEgressConfigRejectedAtBuildWithMinZero() { } @Test - public void testAutoFlushOffAcceptedAtBuildWithMinZero() { - // auto_flush_interval=off and auto_flush=off are legal on WebSocket. + public void testAutoFlushIntervalOffAcceptedAtBuildWithMinZero() { + // auto_flush_interval=off turns off only the time-based trigger and is legal on + // WebSocket, alone or with the row trigger off too (only auto_flush=off is rejected). QuestDB.builder().fromConfig( "ws::addr=127.0.0.1:1;auto_flush_interval=off;sender_pool_min=0;query_pool_min=0;").build().close(); QuestDB.builder().fromConfig( - "ws::addr=127.0.0.1:1;auto_flush=off;sender_pool_min=0;query_pool_min=0;").build().close(); + "ws::addr=127.0.0.1:1;auto_flush_rows=off;auto_flush_interval=off;auto_flush_bytes=65536;" + + "sender_pool_min=0;query_pool_min=0;").build().close(); } @Test @@ -227,12 +229,18 @@ public void testMalformedIngressConfigRejectedAtBuildWithMinZero() { // Sender -- yet it must still reject a malformed ingress key in the // single cluster config up front. Covers a typed enum (tls_verify), a // registry-STRING value that only the real Sender parse validates - // (auto_flush_rows), and a WebSocket build-time check: sf_durability=flush - // is not yet supported. + // (auto_flush_rows), and WebSocket build-time checks: auto_flush=off disables + // auto-flush (unsupported on WebSocket), also when auto_flush_bytes is set, and + // sf_durability=flush is not yet supported. assertBuildRejected( "wss::addr=127.0.0.1:1;tls_verify=strict;sender_pool_min=0;query_pool_min=0;", "tls_verify"); assertBuildRejected( "ws::addr=127.0.0.1:1;auto_flush_rows=abc;sender_pool_min=0;query_pool_min=0;", "auto_flush_rows"); + assertBuildRejected( + "ws::addr=127.0.0.1:1;auto_flush=off;sender_pool_min=0;query_pool_min=0;", "auto-flush"); + assertBuildRejected( + "ws::addr=127.0.0.1:1;auto_flush=off;auto_flush_bytes=65536;sender_pool_min=0;query_pool_min=0;", + "auto-flush"); assertBuildRejected( "ws::addr=127.0.0.1:1;sf_durability=flush;sender_pool_min=0;query_pool_min=0;", "not yet supported"); } diff --git a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/DeltaDictRecoveryTest.java b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/DeltaDictRecoveryTest.java index 3b8d316c..fbc833d8 100644 --- a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/DeltaDictRecoveryTest.java +++ b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/DeltaDictRecoveryTest.java @@ -1118,6 +1118,8 @@ public void testFailedPublishThenNewSymbolPersistsSuffixWithoutDuplicating() thr // atNow() and throws from there instead of from the flush() this test // brackets. The batch and the branch under test are identical either way; // only the throw site moves, so parking the thresholds costs no coverage. + // (WebSocket rejects auto_flush=off outright, hence the large finite + // interval.) String cfg = "ws::addr=localhost:" + port + ";sf_dir=" + sfDir + ";sf_max_segment_bytes=1024" + ";auto_flush_bytes=off;auto_flush_rows=1000000;auto_flush_interval=60000;"; diff --git a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/LineSenderBuilderWebSocketTest.java b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/LineSenderBuilderWebSocketTest.java index c5dad492..1d83dff2 100644 --- a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/LineSenderBuilderWebSocketTest.java +++ b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/LineSenderBuilderWebSocketTest.java @@ -425,21 +425,12 @@ public void testDefaultIsAsync() { } @Test - public void testDisableAutoFlush_supportedForWebSocket() throws Exception { - assertMemoryLeak(() -> { - try (TestWebSocketServer server = new TestWebSocketServer(new TestWebSocketServer.WebSocketServerHandler() { - })) { - server.start(); - Assert.assertTrue(server.awaitStart(5, TimeUnit.SECONDS)); - try (QwpWebSocketSender sender = (QwpWebSocketSender) Sender.builder(Sender.Transport.WEBSOCKET) - .address(LOCALHOST + ":" + server.getPort()) - .disableAutoFlush() - .build()) { - Assert.assertEquals(0, sender.getAutoFlushRows()); - Assert.assertEquals(0L, sender.getAutoFlushIntervalNanos()); - } - } - }); + public void testDisableAutoFlush_notSupportedForWebSocket() { + assertThrowsAny( + Sender.builder(Sender.Transport.WEBSOCKET) + .address(LOCALHOST) + .disableAutoFlush(), + "not supported for WebSocket"); } @Test diff --git a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/SymbolDictRecycleFsnContinuityTest.java b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/SymbolDictRecycleFsnContinuityTest.java index 93032ae6..76c33bb5 100644 --- a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/SymbolDictRecycleFsnContinuityTest.java +++ b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/SymbolDictRecycleFsnContinuityTest.java @@ -392,7 +392,8 @@ public void testLatchedErrorStillThrowsForOldEpochTarget() throws Exception { * "greater than" check but fails the exact equalities below. Arming uses * resetSymbolDictionary() so the test does not depend on threshold or * anti-thrash floor arithmetic; auto-flush stays off so no interval - * flush can inject frames on a slow CI box. This pins the same + * flush can inject frames on a slow CI box. WebSocket rejects + * {@code auto_flush=off} outright, so this pins the same * large-finite-interval idiom already * used by {@code DeltaDictRecoveryTest} / {@code SelfSufficientFramesTest} * to fully suppress auto-flush on WS. diff --git a/core/src/test/java/io/questdb/client/test/impl/WsSenderConfigHonoredTest.java b/core/src/test/java/io/questdb/client/test/impl/WsSenderConfigHonoredTest.java index e4b94d8f..39695ec2 100644 --- a/core/src/test/java/io/questdb/client/test/impl/WsSenderConfigHonoredTest.java +++ b/core/src/test/java/io/questdb/client/test/impl/WsSenderConfigHonoredTest.java @@ -54,7 +54,8 @@ public void testEveryIngressKeyIsHonored() { assertHonored("auto_flush_interval=250", "auto_flush_interval", 250); // auto_flush=off reaches disableAutoFlush(), which sets the interval to // MAX_VALUE; auto_flush_rows=off leaves the interval unset, so this field - // distinguishes the two. + // distinguishes the two. That config is build-rejected on WebSocket; see + // QuestDBBuilderTest.testMalformedIngressConfigRejectedAtBuildWithMinZero. assertHonored("auto_flush=off", "auto_flush_interval", Integer.MAX_VALUE); assertHonored("max_name_len=99", "max_name_len", 99); assertHonored("transaction=on", "transaction", true); From 1da8ffd466a4332b7e25d59f9e681844dcf37fa1 Mon Sep 17 00:00:00 2001 From: Jaromir Hamala Date: Thu, 5 Nov 2026 20:14:15 +0100 Subject: [PATCH 13/13] Remove unused sendBinary override in test stub BlockingDoorbellClient overrode the one-slice sendBinary to append and flush a single frame, but the send loop no longer calls that overload: data frames go through batchAppendBinary and batchFlush, and the dictionary catch-up uses the two-slice sendBinary. Drop the dead override and its comment. --- .../sf/cursor/CursorWebSocketSendLoopIdleWaitTest.java | 7 ------- 1 file changed, 7 deletions(-) diff --git a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoopIdleWaitTest.java b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoopIdleWaitTest.java index f55ad33f..6196b197 100644 --- a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoopIdleWaitTest.java +++ b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoopIdleWaitTest.java @@ -516,13 +516,6 @@ public void batchFlush() { getSendBuffer().reset(); } - // The one-frame-per-write path: one frame, one write. - @Override - public void sendBinary(long dataPtr, int length) { - batchAppendBinary(dataPtr, length); - batchFlush(); - } - @Override public void sendPing() { pingAttempts.incrementAndGet();