diff --git a/core/src/main/java/io/questdb/client/Sender.java b/core/src/main/java/io/questdb/client/Sender.java index d0dab97c..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; @@ -1865,7 +1869,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. + * 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; } @@ -3561,9 +3567,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,7 +5068,7 @@ 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) { + if (autoFlushDisabled) { throw new LineSenderException("disabling auto-flush is not supported for WebSocket protocol"); } if ((sfDurability == SfDurability.FLUSH || sfDurability == SfDurability.APPEND)) { 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..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 @@ -241,6 +241,42 @@ 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 #batchByteCap()} 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); + } + + /** + * 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 batchByteCap() { + return DEFAULT_SEND_BUFFER_SIZE; + } + + /** + * 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 @@ -483,6 +519,7 @@ public boolean receiveFrame(WebSocketFrameHandler handler, int timeout) { } // Ensure buffer has space + compactRecvBuffer(); if (recvPos >= recvBufSize - 1024) { growRecvBuffer(); } @@ -668,6 +705,7 @@ public boolean tryReceiveFrame(WebSocketFrameHandler handler) { } // Try one non-blocking recv + compactRecvBuffer(); if (recvPos >= recvBufSize - 1024) { growRecvBuffer(); } @@ -1145,18 +1183,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); } } } @@ -1414,8 +1457,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/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/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/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..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 @@ -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 @@ -2453,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; @@ -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. @@ -2908,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()}. */ @@ -2916,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() { @@ -7132,8 +7145,9 @@ private void sendRow() { long bufferedNow = currentTableBuffer.nextRow( currentTableBufferSnapshotBytes, cap > 0 ? cap : Long.MAX_VALUE); - if (pendingRowCount == 0) { - firstPendingRowTimeNanos = System.nanoTime(); + // Interval off (0): no clock read on the row path at all. + if (pendingRowCount == 0 && autoFlushIntervalNanos > 0) { + firstPendingRowTimeNanos = nanoClock.getAsLong(); } pendingRowCount++; @@ -7164,8 +7178,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/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..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 @@ -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) @@ -1681,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(); @@ -2760,15 +2762,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 +2796,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 +2916,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); @@ -3108,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) { @@ -3127,7 +3130,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 +3273,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 -- @@ -3339,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() @@ -3375,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; @@ -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,215 @@ 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 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 + * 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() { + // 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; + 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.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 + // 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) { + nextWireSeq = batchStartWireSeq; + replayTargetFsn = batchStartReplayTargetFsn; + 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) { + nextWireSeq = batchStartWireSeq; + replayTargetFsn = batchStartReplayTargetFsn; 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 +4347,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 +4356,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 +4666,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/QuestDBBuilderTest.java b/core/src/test/java/io/questdb/client/test/QuestDBBuilderTest.java index e338aa72..e304df95 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,35 @@ public void testMalformedEgressConfigRejectedAtBuildWithMinZero() { "compression_level"); } + @Test + 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_rows=off;auto_flush_interval=off;auto_flush_bytes=65536;" + + "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 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_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;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/http/client/WebSocketClientTest.java b/core/src/test/java/io/questdb/client/test/cutlass/http/client/WebSocketClientTest.java index d0335886..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 @@ -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; @@ -47,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 @@ -130,6 +189,16 @@ public void testExtractMaxBatchSizeParsesPositive() throws Exception { Assert.assertEquals(16 * 1024 * 1024, invokeExtractMaxBatchSize(response)); } + @Test + public void testFrameBurstIsParsedInPlace() throws Exception { + assertFrameBurstIsParsedInPlace(false); + } + + @Test + public void testFrameBurstIsParsedInPlaceWithBlockingReceive() throws Exception { + assertFrameBurstIsParsedInPlace(true); + } + /** * A frame handler may close() the client from inside its callback: * CursorWebSocketSendLoop's NACK-recycle path (handleServerRejection / @@ -349,6 +418,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 +494,94 @@ 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)) { + 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 +672,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 +708,9 @@ private static class CapturingWebSocketClient extends TestWebSocketClient { @Override protected void ioWait(int timeout, int op) { + if (op == IOOperation.WRITE) { + writeWaits++; + } } @Override @@ -596,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 { 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..b2522295 --- /dev/null +++ b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/AutoFlushIntervalClockTest.java @@ -0,0 +1,165 @@ +/*+***************************************************************************** + * ___ _ ____ ____ + * / _ \ _ _ ___ ___| |_| _ \| __ ) + * | | | | | | |/ _ \/ __| __| | | | _ \ + * | |_| | |_| | __/\__ \ |_| |_| | |_) | + * \__\_\\__,_|\___||___/\__|____/|____/ + * + * 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 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(() -> { + 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 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(() -> { + 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; + } +} 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..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 @@ -393,9 +393,8 @@ public void testLatchedErrorStillThrowsForOldEpochTarget() throws Exception { * 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 + * {@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/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..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 @@ -217,6 +217,69 @@ public void testReplayAfterPingFailureIsNotStrandedByIdleWait() throws Exception }); } + /** + * 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 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.batchByteCap() / wireLen; + assertTrue("test needs several frames per write", framesPerWrite > 1 && framesPerWrite < frames); + 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. + */ + @Test + public void testBurstOfPublishedFramesGoesOutInOneWrite() throws Exception { + TestUtils.assertMemoryLeak(() -> { + try (CursorSendEngine engine = new CursorSendEngine(tmpDir, 16384)) { + BlockingDoorbellClient client = new BlockingDoorbellClient(); + publishBurstToHeldLoop(engine, client, 0, 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 +328,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 +355,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 +388,56 @@ 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 (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 leadPayloadLen, 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)); + 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() < total && System.nanoTime() < deadline) { + Thread.sleep(1); + } + assertEquals("not every published frame was sent", total, loop.getTotalFramesSent()); + assertEquals(total, 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 +448,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,9 +459,11 @@ 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; + // Largest send buffer capacity seen at a flush. + volatile int maxBufferCapacity; private boolean okDelivered; private int receivesSinceSend; private final Semaphore doorbell = new Semaphore(0); @@ -381,15 +497,25 @@ 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]"); } + // Through the real send buffer, so a large frame grows it. + getSendBuffer().appendMaskedBinaryFrame(dataPtr, length); sentFrames.incrementAndGet(); receivesSinceSend = 0; } + // One socket write per flush, like the real client. + @Override + public void batchFlush() { + writes.incrementAndGet(); + maxBufferCapacity = Math.max(maxBufferCapacity, getSendBuffer().getCapacity()); + getSendBuffer().reset(); + } + @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/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; + } + } +} 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..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,8 +43,8 @@ import org.junit.Test; /** - * Probabilistic regression test for the send-loop rotation race fixed in - * {@code CursorWebSocketSendLoop.trySendOne()} (PR #41): the loop could observe + * 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. *

@@ -85,7 +87,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( @@ -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 sendBinary} 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) { @@ -153,10 +216,14 @@ private static final class RecordingClient extends TestWebSocketClient { } @Override - public void sendBinary(long dataPtr, int length) { - if (recordedCount < recorded.length) { - recorded[recordedCount++] = Unsafe.getUnsafe().getLong(dataPtr); - } + public void batchAppendBinary(long dataPtr, int length) { + Assert.assertTrue("more frames than expected", appendedCount < recorded.length); + recorded[appendedCount++] = Unsafe.getUnsafe().getLong(dataPtr); + } + + @Override + public void batchFlush() { + recordedCount = appendedCount; sentCount = recordedCount; } 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) { }