[jerrinot]

Problem

A QWP sender burns ~70% of a core per connection even when idle: the I/O thread polls the socket and the segment ring every 50µs. Kafka Connect users see 30% CPU on QWP vs 3% on HTTP for the same quiet sink.

Change

The I/O thread now blocks until there is something to do: socket readable, a producer published a frame, close(), or a PING is due. The producer wakes it through a native doorbell (eventfd on Linux, kqueue EVFILT_USER on macOS/FreeBSD, kernel Event on Windows) and only when the thread is actually parked, so a busy connection never pays a syscall per flush. Ack waiters, the segment manager and the dispatchers are event-driven as well: a waiter in awaitAckedFsn/drain is unparked by the engine on every ack advance, terminal and durability latch; the segment manager parks until a rotation, ACK or timer needs it and persists the ack watermark on a 1 ms cadence while ACKs stream; the progress dispatcher thread is not started at all unless a handler is installed.

No change to the Sender API or configuration.

Numbers

Single sender, one connection, memory mode, in-JVM loopback server acking every frame. Client CPU is the I/O thread, the segment manager and the producer's time inside flush(), as % of one core; latency is flush-to-ack p50; main = 82b2e144, PR = this branch. Medians of 3 interleaved 5 s runs.

markbook (i7-4870HQ, 4C/8T)

flushes/s	CPU main	CPU PR	ack latency main	ack latency PR
idle	20.3%	0%
1k	8.4%	4.4%	171 µs	100 µs
10k	24.2%	22.7%	170 µs	30 µs
25k	43.0%	49.5%	131 µs	27 µs
50k	75.9%	79.1%	99 µs	33 µs
saturation	107k fl/s	134k fl/s

domapc (Ryzen 9 9950X, 16C/32T)

flushes/s	CPU main	CPU PR	ack latency main	ack latency PR
idle	1.8%	0%
1k	2.0%	0.7%	149 µs	11 µs
10k	6.0%	5.8%	171 µs	10 µs
25k	10.7%	12.8%	106 µs	7 µs
50k	18.8%	25.4%	82 µs	7 µs
saturation	400k fl/s	477k fl/s

Trade-off: above ~25k small flushes per second per connection the client uses more CPU than main (1.1–1.45x), because the I/O thread now takes two prompt wakes per flush (doorbell, then ACK) where the old loop batched several flushes per 50 µs park. In exchange, flush-to-ack latency drops 3–15x, saturation throughput rises ~20%, and idle costs nothing. With the default auto_flush_rows=1000 this band starts around 25M rows/s per connection; it matters for explicit per-event flushes.

Kafka Connect (4 connections, 1 row/s): 69% → 1.9% of a core, HTTP 1.7%.

Notes for review

• design/tla holds TLA+ models of the two wake protocols; IdleWait_bug.cfg reproduces the reconnect race the models caught.
• macOS and Windows natives compile only in CI.
• Server side: durable-ack frames are still only flushed on inbound traffic, so the 200ms durable-ack PING stays until the server pushes them.
• The native wait entry point is Net.awaitReadableOrRing: it consumes the doorbell ring itself, so a stale local libquestdb.so fails with UnsatisfiedLinkError instead of spinning.

[jerrinot]

Pushed a fix for the ack-watermark regression found in review (39ac01e).

Root cause: the mmap watermark write ran on the manager's 1 ms tick. With the worker event-driven, only an ACK that completed a sealed segment woke it, so ACKs inside the active segment were never persisted. After kill -9, recovery replayed them.

Fix: the first ACK past the persisted watermark wakes the worker; it then re-persists on a 1 ms cadence (WATERMARK_LAG_NANOS) until the watermark is current, then parks again. Same bound as before, no polling when idle. New regression test SegmentManagerAckWatermarkLagTest fails on the previous commit and passes now. The three SenderPoolSfTest tests that assumed the tick now wake the worker explicitly.

Full suite: 3725 tests, 0 failures.

[jerrinot]

Pushed a fix for the lost-waiter finding (awaitAckedFsn / drain sleeping to their deadline).

Root cause: one ackWaiter slot on the I/O loop. A second waiter overwrote the first, and a waiter parked on a loop that a symbol-dictionary recycle closed had no wake source left.

Fix: waiters register in a Sender-owned AckWaiters registry (any number of threads, allocation-free signal) and every loop the sender creates signals it on ack advance, terminal latch and close(). Register-then-check against advance-then-signal, same hand-off the TLA+ model checks.

testAwaitAckedFsnParkedAcrossARecycleSeesTheFreshEpochAck is back from 10 s to sub-second and now asserts the monitor returns within 2 s; new testEveryWaiterIsWokenByTheAck covers three concurrent monitors. Full suite: 3726 tests, 0 failures.

[jerrinot]

Pushed a fix for the stranded-frame schedule (idle re-check vs. the two-store publish).

Root cause: a publish stores publishedFsn first and the bytes second (deliberately, for the ack clamp). The loop can only send below the bytes, but its idle re-check keyed on publishedFsn. Your schedule: snapshot the new FSN, find no bytes, producer finishes the publish and sees ioParked == false, loop re-checks publishedFsn, unchanged, blocks with no timeout.

Fix: SegmentRing bumps a publishStamp after the bytes store; the loop snapshots and re-checks that. The Dekker argument now covers the last store of the publish.

The TLA+ model splits the publish into both stores. IdleWait_stampbug.cfg (re-check on publishedFsn) reproduces exactly this schedule; IdleWait_fix.cfg passes all invariants and liveness. Full suite: 3726 tests, 0 failures.

[jerrinot]

Review (level 3): request changes

HEAD 5d7f6bab vs BASE 82b2e144. No submodule pointers moved. Every behavioural finding below was reproduced on both revisions with the same trigger; reproduction sources are at the end.

Critical

1. TLS: an ACK frame larger than ~64 KiB is left in the TLS layer's user-space buffers, and the I/O loop then blocks with no timeout

• Problem: Over wss, a large server frame can sit in JavaTlsClientSocket's buffers while the I/O loop waits on the raw socket.
• Net impact: On Enterprise TLS, the ACK is never processed until another publish. drain()/awaitAckedFsn time out and close() fails.
• Evidence: Repro test below:
  • awaitAckedFsn: with a single 70 KiB or 150 KiB STATUS_OK frame, HEAD returned acked=false after 5000 ms in 6/6 runs. BASE acked in 3–37 ms in 6/6.
  • close() (with close_flush_timeout_millis=3000): HEAD threw "close() drain timed out … data may be lost" in 5/6 runs. BASE closed cleanly in 6/6.
  • Controls: plain ws passed at every size. TLS frames of 16–60 KiB passed.

Trace

1. tryReceiveFrame makes one socket.recv call.
2. JavaTlsClientSocket.recv reads up to 256 KiB of ciphertext. It returns on BUFFER_OVERFLOW when the 64 KiB receive buffer fills, or it spills plaintext into unwrapOutputBuffer. Either way, the rest of the frame stays in user space.
3. The frame is still incomplete, so tryReceiveFrame returns false and tryReceiveAcks stops.
4. awaitIdle then polls the kernel fd, which is empty. Keepalive is off by default, so the timeout is -1.

On BASE the 50 µs park simply called recv again.

Who hits it

• On the server, QwpIngressUpgradeProcessor.trySendAck writes one entry (nameLen + name + seqTxn) for every table committed since the last ACK, which is sent every ≤8 frames. Durable-ack frames have the same table-list layout.
• Enterprise serves QWP over TLS (http.tls.enabled).
• So a wss sender that commits roughly 2,000+ distinct tables in one ACK window and then goes idle hits this. One example is the last batch before close().

Fix (validated) In WebSocketClient.tryReceiveFrame, keep calling recv while it returns > 0 and the frame is still PARSE_NEED_MORE. Alternatively, treat "bytes received" as didWork in the loop.

• With the patch below, the repro acks in 1–7 ms at every size.
• WebSocketClientTest, WebSocketClientDoorbellTest and CursorWebSocketSendLoopIdleWaitTest still pass.
• Please add a TLS regression test along the lines of the repro.

Validated fix (WebSocketClient.tryReceiveFrame)

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 68b2c2d0..d8ed020d 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
@@ -640,18 +640,25 @@ public abstract class WebSocketClient implements QuietCloseable {
             growRecvBuffer();
         }
 
-        int n = socket.recv(recvBufPtr + recvPos, recvBufSize - recvPos);
-        if (n < 0) {
-            throw new HttpClientException("peer disconnect [errno=").errno(nf.errno()).put(']');
+        while (true) {
+            int n = socket.recv(recvBufPtr + recvPos, recvBufSize - recvPos);
+            if (n < 0) {
+                throw new HttpClientException("peer disconnect [errno=").errno(nf.errno()).put(']');
+            }
+            if (n == 0) {
+                return false; // No data available
+            }
+            recvPos += n;
+            result = tryParseFrame(handler);
+            if (result != PARSE_NEED_MORE) {
+                return result == PARSE_OK;
+            }
+            // Frame still incomplete: keep reading until the transport has
+            // nothing more (TLS may hold already-received bytes in user space).
+            if (recvPos >= recvBufSize - 1024) {
+                growRecvBuffer();
+            }
         }
-        if (n == 0) {
-            return false; // No data available
-        }
-        recvPos += n;
-
-        // Try to parse again
-        result = tryParseFrame(handler);
-        return result == PARSE_OK;
     }
 
     /**

Moderate

2. A failed segment trim is never retried while the slot is idle

• Problem: The manager's park deadline ignores trimRetryAtNanos.
• Net impact: On disk SF with default durability (sf_durability=memory), acked segments and their budget charge are held until the next ACK or rotation.
• Evidence: One transient failure, then 3 s with no wake (repro below):
  • HEAD: fileExists=true pending=1 for unlink, ackedStillSealed=true for fsyncDir; 3/3 runs.
  • BASE: retried and removed; 3/3 runs.

recordTrimFailure sets a 4 ms–1 s backoff. The workerLoop park computation only folds in the provisioning retry, the periodic-sync deadline and the watermark lag. With the watermark already persisted, the worker parks indefinitely. The existing trim-retry tests call wakeWorker() explicitly, so they can't see this.

• Fix: when e.trimRetryDelayNanos != 0, include e.trimRetryAtNanos - now in parkNanos.

3. A reconnect triggered by a failed PING leaves the replay unsent

• Problem: recycledThisIteration is checked before steps 3–4 (sendPingIfDue) but not before the step-5 park.
• Net impact:
  • Durable-ack mode: the replay waits until the next publish.
  • With keepalive on: the replay waits one keepalive interval.
• Evidence: Repro below:
  • HEAD, durable: replayedOnReplacement=0 after 5 s.
  • HEAD, keepalive=1 s: replayed after 1000 ms.
  • BASE: 0–2 ms in both modes.

Trace

1. sendPing throws, which calls fail → swapClient. The cursor is rewound and pendingDurable is cleared.
2. Step 5 parks. idleTimeoutMillis() is -1 because pendingDurable is now empty.

The trigger is narrow: the PING write itself has to fail, for example a send timeout or an RST that races the PING. But once it fires, the stall in durable mode is unbounded.

• Fix: add !recycledThisIteration to the step-5 guard.

4. The new replay regression test passes with the fix removed

• Problem: testReplayAfterReconnectIsNotStrandedByIdleWait stays green with recycledThisIteration = true deleted from swapClient.
• Net impact: A regression that strands rewound replays would ship undetected.
• Evidence: Mutation run on HEAD:
  • CursorWebSocketSendLoopIdleWaitTest is 2/2 green with the line removed.
  • A probe whose first sendBinary throws replays in 3 ms with the flag and stays stranded for 3 s without it (testSendFailureReplay below).

The test's transport fails on receive after a successful send, so didWork is already true in that iteration. It should fail on sendBinary (or sendPing) instead, with the producer idle.

5. AckWaiters allocates on every wait iteration, and every ACK triggers a new iteration

• Problem: add calls Arrays.copyOf and remove calls new Thread[n-1] on every loop pass in awaitAckedFsn/drainOnClose. remove allocates a fresh Thread[0] instead of reusing NONE.
• Net impact: drain()/awaitAckedFsn allocate about 2 arrays per ACK received while they wait. BASE allocated nothing.
• Evidence: Static: AckWaiters.java:42-58. The arrays are published through a volatile field, so escape analysis can't remove them. signal() wakes every waiter on each ACK advance.

This is per call, not per row, but it is still garbage on a flush-and-wait API.

• Fix:
  • Register once per call, outside the loop. The unpark permit already covers the gap between check and park.
  • Use a slot array that only grows when needed, so steady-state waits allocate nothing.

6. The documented binary-compatibility constructor no longer exists, and SegmentManager(long,long) changed meaning

• Problem: CursorWebSocketSendLoop's overload documented as "as released in 1.3.x … kept so callers compiled against a released jar keep linking" dropped parkNanos.
• Net impact: Code compiled against 1.3.x that calls these exported sf.cursor constructors either fails to link or misbehaves (details below). I found no callers in the questdb or questdb-enterprise sources.
• Evidence: Static:
  • Tags 1.3.7–1.3.9 have (…, long fsnAtZero, long parkNanos, …) and SegmentManager(long segmentSizeBytes, long pollNanos).
  • HEAD has SegmentManager(long segmentSizeBytes, long maxTotalBytes), which has the same (JJ)V descriptor.

Old callers of the loop constructors get NoSuchMethodError. An old new SegmentManager(seg, 1_000_000L) links fine and is then read as a 1 MB cap: it throws IAE if the segment is larger than 1 MB, and silently caps the budget otherwise.

• Fix:
  • Keep @Deprecated overloads with the old descriptors that ignore parkNanos.
  • Don't reuse (long,long) with a new meaning.

Minor

• SegmentRing.java:1507-1521: the publishedFsn() javadoc now sits above publishStamp()'s javadoc. That leaves it dangling, and publishedFsn() has no javadoc.
• These comments are stale:
  • the awaitAckedFsn javadoc ("Polls on a 50us park"; "observed on the next pass");
  • SenderErrorDispatcher.close() ("100ms poll timeout");
  • design/tla/AckWait.tla, which models a CursorWebSocketSendLoop.awaitAckProgress()/single-waiter design that doesn't exist (the code uses AckWaiters);
  • the CursorWebSocketSendLoopRotationRaceTest javadoc, which still describes 1 µs polling.
• Member ordering:
  • SegmentManager: PROVISION_RETRY_NANOS and WATERMARK_LAG_NANOS come before DISK_FULL_LOG_THROTTLE_NANOS, and provisionRetryDeadlineNanos sits among the final fields.
  • CursorSendEngine.ioWakeup sits among the final fields.
  • The AckWaiters import in QwpWebSocketSender is out of order.

Coverage and CI

• Test gate: passes, with no Critical coverage gap. The tests missing for findings 1–3 are listed in their fixes.
• Tandem: No OSS or Enterprise tandem PR exists (gh pr list --head qwp-event-driven-io-loop returned nothing in both repos). Finding 1 is exactly the kind of thing an Enterprise TLS e2e run with many tables would exercise.
• CI:
  • Everything is green, including the Azure macOS and Windows build-and-test legs.
  • The exception is the linux-aarch64 GLIBC floor guard. It failed in "Install tooling": the EPEL mirror returned a certificate error and a checksum mismatch on repomd.xml, before any build step ran. That is an infrastructure failure, but it means the new native code is unchecked on aarch64. Please re-run it.

Summary

• Verdict: request changes, because finding 1 is Critical.
• Split: 1 Critical, 5 Moderate and a few Minors. All are in-diff.
• Tradeoffs: The idle-CPU win is real. Two I/O-thread ACK-path costs are possible but unmeasured, so they are not filed here:
  • the I/O thread now takes the ring monitor via firstTrimmable();
  • the manager wakes once per ACK under load.

Repro: TLS large-ACK stall (finding 1), core/src/test/java/io/questdb/client/test/cutlass/qwp/client/TlsLargeAckStallScratchTest.java

package io.questdb.client.test.cutlass.qwp.client;

import io.questdb.client.Sender;
import io.questdb.client.test.cutlass.qwp.websocket.TestWebSocketServer;
import org.junit.Assert;
import org.junit.Test;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.ByteOrder;
import java.nio.charset.StandardCharsets;
import java.util.Collections;
import java.util.IdentityHashMap;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;

public class TlsLargeAckStallScratchTest {

    @Test(timeout = 180_000)
    public void testBigAck150k() throws Exception { for (int i = 0; i < 3; i++) runOnce(150 * 1024, i); }

    @Test(timeout = 180_000)
    public void testBigAck70k() throws Exception { for (int i = 0; i < 3; i++) runOnce(70 * 1024, i); }

    private static void runOnce(int bigSize, int run) throws Exception {
        BigFirstAckHandler handler = new BigFirstAckHandler(bigSize);
        try (TestWebSocketServer server = TestWebSocketServer.tls(handler)) {
            server.start();
            Assert.assertTrue(server.awaitStart(5, TimeUnit.SECONDS));
            try (Sender sender = Sender.fromConfig("wss::addr=localhost:" + server.getPort() + ";tls_verify=unsafe_off;")) {
                sender.table("t").longColumn("v", 1).atNow();
                long fsn = sender.flushAndGetSequence();
                long t0 = System.nanoTime();
                boolean acked = sender.awaitAckedFsn(fsn, 5_000);
                long waitMs = (System.nanoTime() - t0) / 1_000_000L;
                System.out.println("RESULT run=" + run + " size=" + bigSize + " ackPayload=" + handler.bigPayloadLen
                        + " acked=" + acked + " waitMs=" + waitMs + " framesAtServer=" + handler.frames.get()
                        + " fsn=" + fsn + " ackedFsn=" + sender.getAckedFsn());
                // Unrelated event: a second publish should unstick a stalled loop.
                sender.table("t").longColumn("v", 2).atNow();
                long fsn2 = sender.flushAndGetSequence();
                long t1 = System.nanoTime();
                boolean acked2 = sender.awaitAckedFsn(fsn2, 5_000);
                System.out.println("RESULT2 run=" + run + " size=" + bigSize + " ackedAfterSecondPublish=" + acked2
                        + " waitMs=" + (System.nanoTime() - t1) / 1_000_000L + " ackedFsn=" + sender.getAckedFsn());
            }
        }
    }

    static byte[] buildBigAck(long seq, int targetPayload) {
        int header = 1 + 8 + 2;
        int entry = 2 + 8 + 8; // nameLen + "t_NNNNNN" + seqTxn
        int count = (targetPayload - header + entry - 1) / entry;
        ByteBuffer bb = ByteBuffer.allocate(header + count * entry).order(ByteOrder.LITTLE_ENDIAN);
        bb.put((byte) 0x00).putLong(seq).putShort((short) count);
        for (int i = 0; i < count; i++) {
            byte[] name = String.format("t_%06d", i).getBytes(StandardCharsets.US_ASCII);
            bb.putShort((short) name.length).put(name).putLong(1L);
        }
        return bb.array();
    }

    static final class BigFirstAckHandler implements TestWebSocketServer.WebSocketServerHandler {
        final int bigSize;
        final AtomicLong frames = new AtomicLong();
        volatile int bigPayloadLen;
        private final Map<TestWebSocketServer.ClientHandler, AtomicLong> seqs =
                Collections.synchronizedMap(new IdentityHashMap<TestWebSocketServer.ClientHandler, AtomicLong>());

        BigFirstAckHandler(int bigSize) { this.bigSize = bigSize; }

        @Override
        public void onBinaryMessage(TestWebSocketServer.ClientHandler client, byte[] data) {
            frames.incrementAndGet();
            AtomicLong s = seqs.get(client);
            if (s == null) { s = new AtomicLong(); seqs.put(client, s); }
            long seq = s.getAndIncrement();
            byte[] ack = seq == 0 ? buildBigAck(seq, bigSize) : WebSocketDynamicCredentialTest.buildAck(seq);
            if (seq == 0) bigPayloadLen = ack.length;
            try { client.sendBinary(ack); } catch (IOException e) { throw new RuntimeException(e); }
        }
    }
}

Repro: idle trim retry (finding 2), sf/cursor/ScratchTrimIdleRetryTest.java (BASE: use the commented DEFAULT_POLL_NANOS ctor)

package io.questdb.client.test.cutlass.qwp.client.sf.cursor;

import io.questdb.client.cutlass.qwp.client.sf.cursor.AckWatermark;
import io.questdb.client.cutlass.qwp.client.sf.cursor.MmapSegment;
import io.questdb.client.cutlass.qwp.client.sf.cursor.SegmentManager;
import io.questdb.client.cutlass.qwp.client.sf.cursor.SegmentRing;
import io.questdb.client.std.Files;
import io.questdb.client.std.FilesFacade;
import io.questdb.client.std.MemoryTag;
import io.questdb.client.std.Unsafe;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;

import java.nio.file.Paths;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

public class ScratchTrimIdleRetryTest {
    private static final String REV = "HEAD"; // "BASE" in the base worktree
    private String tmpDir;

    private static SegmentManager newManager(long seg, FilesFacade ff) {
        return new SegmentManager(seg, seg * 8, ff);                                   // HEAD: System::nanoTime, production-equivalent
        // return new SegmentManager(seg, SegmentManager.DEFAULT_POLL_NANOS, seg * 8, ff); // BASE: 1 ms poll as in CursorSendEngine, System::nanoTime
    }

    @Before
    public void setUp() {
        tmpDir = Paths.get(System.getProperty("java.io.tmpdir"), "qdb-scratch-trim-" + System.nanoTime()).toString();
        Assert.assertEquals(0, Files.mkdir(tmpDir, Files.DIR_MODE_DEFAULT));
    }

    @After
    public void tearDown() { removeRecursive(tmpDir); }

    @Test public void unlinkOnce() throws Exception { run(false); }
    @Test public void fsyncDirOnce() throws Exception { run(true); }

    private void run(boolean fsyncFault) throws Exception {
        long seg = MmapSegment.HEADER_SIZE + MmapSegment.FRAME_HEADER_SIZE + 32L;
        String dir = tmpDir + (fsyncFault ? "/fsync" : "/unlink");
        Assert.assertEquals(0, Files.mkdir(dir, Files.DIR_MODE_DEFAULT));
        String acked = dir + "/sf-0000000000000000.sfa";
        SegmentRing ring = new SegmentRing(MmapSegment.create(acked, 0L, seg), seg);
        AckWatermark wm = AckWatermark.open(dir);
        Assert.assertNotNull(wm);
        wm.write(-1L);
        long payload = Unsafe.malloc(32, MemoryTag.NATIVE_DEFAULT);
        try {
            Assert.assertEquals(0L, ring.appendOrFsn(payload, 32));
            ring.installHotSpare(MmapSegment.create(dir + "/sf-0000000000000001.sfa", 1L, seg));
            Assert.assertEquals(1L, ring.appendOrFsn(payload, 32));
            // Idle steady state: spare already provisioned -> no provision fsync / provision-retry timer.
            ring.installHotSpare(MmapSegment.create(dir + "/sf-0000000000000002.sfa", 2L, seg));
            ring.acknowledge(0L);
            OneShotFacade ff = new OneShotFacade(fsyncFault ? null : acked, fsyncFault ? dir : null);
            try (SegmentManager m = newManager(seg, ff)) {
                m.register(ring, dir, wm);
                ff.armed = true;
                m.start();
                Assert.assertTrue("fault never hit", ff.firstFault.await(5, TimeUnit.SECONDS));
                Thread.sleep(3_000L); // no ACK, no append, no wakeWorker()
                MmapSegment fs = ring.firstSealed();
                System.out.println("SCRATCH_RESULT rev=" + REV + " kind=" + (fsyncFault ? "fsyncDir" : "unlink")
                        + " faults=" + ff.faults.get() + " removeCalls=" + ff.removeCalls.get()
                        + " fsyncDirCalls=" + ff.fsyncDirCalls.get() + " fileExists=" + Files.exists(acked)
                        + " pending=" + ring.getPendingTrimCount()
                        + " ackedStillSealed=" + (fs != null && acked.equals(fs.path()))
                        + " managerBytes=" + m.getTotalBytesForTesting() + " ringBytes=" + ring.totalSegmentBytes()
                        + " segSize=" + seg);
            }
        } finally {
            Unsafe.free(payload, 32, MemoryTag.NATIVE_DEFAULT);
            ring.close();
            wm.close();
        }
    }

    private static void removeRecursive(String dir) {
        long find = Files.findFirst(dir);
        if (find > 0) {
            try {
                int rc = 1;
                while (rc > 0) {
                    String name = Files.utf8ToString(Files.findName(find));
                    if (name != null && !".".equals(name) && !"..".equals(name)) {
                        String child = dir + "/" + name;
                        if (!Files.remove(child)) removeRecursive(child);
                    }
                    rc = Files.findNext(find);
                }
            } finally { Files.findClose(find); }
        }
        Files.remove(dir);
    }

    // Delegating facade; method set mirrors FailingFilesFacade in SegmentManagerUnlinkFailureTest of the same revision.
    private static final class OneShotFacade implements FilesFacade {
        final String removeFailPath; final String fsyncFailDir;
        final CountDownLatch firstFault = new CountDownLatch(1);
        final AtomicInteger faults = new AtomicInteger(), removeCalls = new AtomicInteger(), fsyncDirCalls = new AtomicInteger();
        volatile boolean armed;
        OneShotFacade(String removeFailPath, String fsyncFailDir) { this.removeFailPath = removeFailPath; this.fsyncFailDir = fsyncFailDir; }
        @Override public boolean remove(String path) {
            if (path.equals(removeFailPath)) {
                removeCalls.incrementAndGet();
                if (armed && faults.get() == 0) { faults.incrementAndGet(); firstFault.countDown(); return false; }
            }
            return INSTANCE.remove(path);
        }
        @Override public int fsyncDir(String d) {
            if (d.equals(fsyncFailDir) && armed) {
                fsyncDirCalls.incrementAndGet();
                if (faults.get() == 0) { faults.incrementAndGet(); firstFault.countDown(); return -1; }
            }
            return INSTANCE.fsyncDir(d);
        }
        @Override public boolean allocate(int fd, long size) { return INSTANCE.allocate(fd, size); }
        @Override public long allocNativePath(String p) { return INSTANCE.allocNativePath(p); }
        @Override public int close(int fd) { return INSTANCE.close(fd); }
        @Override public boolean exists(String p) { return INSTANCE.exists(p); }
        @Override public void findClose(long f) { INSTANCE.findClose(f); }
        @Override public long findFirst(String d) { return INSTANCE.findFirst(d); }
        @Override public long findName(long f) { return INSTANCE.findName(f); }
        @Override public int findNext(long f) { return INSTANCE.findNext(f); }
        @Override public int findType(long f) { return INSTANCE.findType(f); }
        @Override public void freeNativePath(long p) { INSTANCE.freeNativePath(p); }
        @Override public int fsync(int fd) { return INSTANCE.fsync(fd); }
        @Override public long length(int fd) { return INSTANCE.length(fd); }
        @Override public long length(String p) { return INSTANCE.length(p); }
        @Override public long length(long p) { return INSTANCE.length(p); }
        @Override public int lock(int fd) { return INSTANCE.lock(fd); }
        @Override public int mkdir(String p, int m) { return INSTANCE.mkdir(p, m); }
        @Override public int openCleanRW(String p) { return INSTANCE.openCleanRW(p); }
        @Override public int openCleanRW(long p) { return INSTANCE.openCleanRW(p); }
        @Override public int openRW(String p) { return INSTANCE.openRW(p); }
        @Override public int openRW(long p) { return INSTANCE.openRW(p); }
        @Override public long read(int fd, long a, long l, long o) { return INSTANCE.read(fd, a, l, o); }
        @Override public boolean remove(long p) { return INSTANCE.remove(p); }
        @Override public int rename(String a, String b) { return INSTANCE.rename(a, b); }
        @Override public boolean truncate(int fd, long s) { return INSTANCE.truncate(fd, s); }
        @Override public long write(int fd, long a, long l, long o) { return INSTANCE.write(fd, a, l, o); }
    }
}

Repro: PING/send-failure replay (findings 3 and 4), sf/cursor/PingFailureReplayProbeTest.java + ProbeLoops.java (BASE ctor takes DEFAULT_PARK_NANOS after fsnAtZero)

package io.questdb.client.test.cutlass.qwp.client.sf.cursor;

import io.questdb.client.DefaultHttpClientConfiguration;
import io.questdb.client.cutlass.http.client.WebSocketClient;
import io.questdb.client.cutlass.http.client.WebSocketFrameHandler;
import io.questdb.client.cutlass.qwp.client.sf.cursor.CursorSendEngine;
import io.questdb.client.cutlass.qwp.client.sf.cursor.CursorWebSocketSendLoop;
import io.questdb.client.network.PlainSocketFactory;
import io.questdb.client.std.Files;
import io.questdb.client.std.MemoryTag;
import io.questdb.client.std.Unsafe;
import org.junit.Test;

import java.nio.file.Paths;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

public class PingFailureReplayProbeTest {

    @Test
    public void testKeepalivePingFailure() throws Exception {
        run(false);
    }

    @Test
    public void testDurableAckPingFailure() throws Exception {
        run(true);
    }

    @Test
    public void testSendFailureReplay() throws Exception {
        String dir = Paths.get(System.getProperty("java.io.tmpdir"), "qdb-send-probe-" + System.nanoTime()).toString();
        Files.mkdir(dir, Files.DIR_MODE_DEFAULT);
        try (CursorSendEngine engine = new CursorSendEngine(dir, 16384)) {
            StubClient first = new StubClient(false, false);
            first.failFirstSend = true;
            StubClient replacement = new StubClient(false, false);
            CursorWebSocketSendLoop loop = ProbeLoops.newLoop(first, engine, () -> replacement, false);
            try {
                loop.start();
                long buf = Unsafe.malloc(16, MemoryTag.NATIVE_DEFAULT);
                try {
                    engine.appendBlocking(buf, 16);
                } finally {
                    Unsafe.free(buf, 16, MemoryTag.NATIVE_DEFAULT);
                }
                long t0 = System.nanoTime();
                long d2 = t0 + TimeUnit.SECONDS.toNanos(3);
                while (replacement.sentFrames.get() == 0 && System.nanoTime() < d2) Thread.sleep(1);
                System.out.println("SEND_PROBE replayedOnReplacement=" + replacement.sentFrames.get()
                        + " delayMs=" + TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - t0));
            } finally {
                loop.close();
            }
        }
    }

    private void run(boolean durable) throws Exception {
        String dir = Paths.get(System.getProperty("java.io.tmpdir"), "qdb-ping-probe-" + System.nanoTime()).toString();
        Files.mkdir(dir, Files.DIR_MODE_DEFAULT);
        try (CursorSendEngine engine = new CursorSendEngine(dir, 16384)) {
            StubClient first = new StubClient(true, durable);
            StubClient replacement = new StubClient(false, false);
            CursorWebSocketSendLoop loop = ProbeLoops.newLoop(first, engine, () -> replacement, durable);
            if (!durable) {
                loop.setKeepaliveIntervalMillis(1_000L);
            }
            try {
                loop.start();
                long buf = Unsafe.malloc(16, MemoryTag.NATIVE_DEFAULT);
                try {
                    for (int i = 0; i < 16; i++) Unsafe.getUnsafe().putByte(buf + i, (byte) 'x');
                    engine.appendBlocking(buf, 16);
                } finally {
                    Unsafe.free(buf, 16, MemoryTag.NATIVE_DEFAULT);
                }
                long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(10);
                while (first.pingAttempts.get() == 0 && System.nanoTime() < deadline) Thread.sleep(1);
                long tFail = System.nanoTime();
                long d2 = tFail + TimeUnit.SECONDS.toNanos(5);
                while (replacement.sentFrames.get() == 0 && System.nanoTime() < d2) Thread.sleep(1);
                System.out.println("PING_PROBE durable=" + durable
                        + " firstSent=" + first.sentFrames.get()
                        + " pingAttempts=" + first.pingAttempts.get()
                        + " replayedOnReplacement=" + replacement.sentFrames.get()
                        + " replayDelayMs=" + TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - tFail)
                        + " ackedFsn=" + engine.ackedFsn());
            } finally {
                loop.close();
            }
        }
    }

    static final class StubClient extends WebSocketClient {
        final AtomicInteger sentFrames = new AtomicInteger();
        final AtomicInteger pingAttempts = new AtomicInteger();
        private final boolean failPing;
        private final boolean deliverDurableOk;
        private final Semaphore doorbell = new Semaphore(0);
        private boolean okDelivered;

        StubClient(boolean failPing, boolean deliverDurableOk) {
            super(DefaultHttpClientConfiguration.INSTANCE, PlainSocketFactory.INSTANCE);
            this.failPing = failPing;
            this.deliverDurableOk = deliverDurableOk;
        }

        // Head-only hooks (no @Override so the file also compiles at base).
        public boolean awaitReadableOrWake(int timeoutMillis) {
            try {
                boolean woken;
                if (timeoutMillis < 0) {
                    doorbell.acquire();
                    woken = true;
                } else {
                    woken = doorbell.tryAcquire(timeoutMillis, TimeUnit.MILLISECONDS);
                }
                doorbell.drainPermits();
                return woken;
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                return false;
            }
        }

        public void wake() {
            doorbell.release();
        }

        volatile boolean failFirstSend;

        @Override
        public void sendBinary(long dataPtr, int length) {
            if (failFirstSend) {
                failFirstSend = false;
                throw new io.questdb.client.cutlass.http.client.HttpClientException("peer disconnect [errno=104]");
            }
            sentFrames.incrementAndGet();
        }

        @Override
        public void sendPing() {
            pingAttempts.incrementAndGet();
            if (failPing) {
                throw new io.questdb.client.cutlass.http.client.HttpClientException("timed out [errno=11]");
            }
        }

        @Override
        public boolean tryReceiveFrame(WebSocketFrameHandler handler) {
            if (deliverDurableOk && !okDelivered && sentFrames.get() > 0) {
                okDelivered = true;
                // STATUS_OK, seq 0, 1 table "t" seqTxn 1
                int len = 1 + 8 + 2 + 2 + 1 + 8;
                long p = Unsafe.malloc(len, MemoryTag.NATIVE_DEFAULT);
                try {
                    Unsafe.getUnsafe().putByte(p, (byte) 0);
                    Unsafe.getUnsafe().putLong(p + 1, 0L);
                    Unsafe.getUnsafe().putShort(p + 9, (short) 1);
                    Unsafe.getUnsafe().putShort(p + 11, (short) 1);
                    Unsafe.getUnsafe().putByte(p + 13, (byte) 't');
                    Unsafe.getUnsafe().putLong(p + 14, 1L);
                    handler.onBinaryMessage(p, len);
                } finally {
                    Unsafe.free(p, len, MemoryTag.NATIVE_DEFAULT);
                }
                return true;
            }
            return false;
        }

        @Override
        protected void ioWait(int timeout, int op) {
        }

        @Override
        protected void setupIoWait() {
        }
    }
}

package io.questdb.client.test.cutlass.qwp.client.sf.cursor;
import io.questdb.client.cutlass.http.client.WebSocketClient;
import io.questdb.client.cutlass.qwp.client.sf.cursor.CursorSendEngine;
import io.questdb.client.cutlass.qwp.client.sf.cursor.CursorWebSocketSendLoop;
final class ProbeLoops {
    static CursorWebSocketSendLoop newLoop(WebSocketClient c, CursorSendEngine e, CursorWebSocketSendLoop.ReconnectFactory f, boolean durable) {
        return new CursorWebSocketSendLoop(c, e, 0L, f, 1L, 10L, durable,
                CursorWebSocketSendLoop.DEFAULT_DURABLE_ACK_KEEPALIVE_INTERVAL_MILLIS,
                CursorWebSocketSendLoop.DEFAULT_MAX_HEAD_FRAME_REJECTIONS);
    }
}

[jerrinot]

Pushed a fix for the TLS large-frame finding.

Root cause: JavaTlsClientSocket holds decrypted plaintext (spill buffer) and complete undecrypted TLS records in user space when a server frame exceeds the caller's buffer. The idle wait polled the fd, which was empty, so a partly received ACK was never finished until the next publish.

Fix: Socket.hasBufferedInput() (default false; the TLS socket reports a non-empty spill buffer or an unprocessed record). awaitReadableOrWake() returns readable without blocking when it is set. The WebSocket layer itself already parses buffered bytes before every recv, so TLS was the only layer hiding progress from poll.

New TlsLargeAckIdleWaitTest sends a 150 KiB STATUS_OK over wss and over plain ws; the TLS variant hangs to its timeout on the previous commit and passes now. Full suite: 3728 tests, 0 failures.

[jerrinot]

Pushed fixes for findings 2 to 6 and the minors, one commit each (98ac6ae, b5e6d52, 05221f2, 898e771, plus the nits commit). Every behavioural fix has a test that fails on the previous commit.

2. Idle trim retry. The manager's park now ends at trimRetryAtNanos too. SegmentManagerUnlinkFailureTest#testFailedUnlinkIsRetriedWithoutExternalWakeup: real clock, one failed unlink, no wakeWorker(); the retry has to come on its own. Fails in 5 s without the fix.

3. PING-failure replay. Step 5 skips the wait when recycledThisIteration was set again by a PING that reconnected. testReplayAfterPingFailureIsNotStrandedByIdleWait (durable-ack mode, PING throws) fails without the guard.

4. Weak replay test. The transport now throws on sendBinary, so the iteration does no work. Mutation-checked: deleting recycledThisIteration = true in swapClient fails this test and the PING one; deleting the step-5 guard fails the PING one only.

5. AckWaiters allocation. Register once per call, loop check/park, deregister on return; remove() reuses the empty array. The park permit covers an ack between check and park; AckWait.tla models the register-once protocol and still passes.

6. 1.3.x descriptors. The seven loop constructors with parkNanos and SegmentManager(long, long pollNanos) / (long, long, long) are back as @Deprecated delegates. (long, long) keeps its old meaning, so a byte cap goes through SegmentBudget (the existing (long, SegmentBudget) overload). Verified with javap that the descriptors are present and, by scanning every call site, that nothing binds to the deprecated ones.

Minors: javadoc re-attached, the four stale comments rewritten, ordering fixed. The aarch64 GLIBC guard passed on the latest run without a rerun.

Full suite: 3730 tests, 0 failures.

[mtopolnik]

[PR Coverage check]

😍 pass : 332 / 376 (88.30%)

file detail

	path	covered line	new line	coverage
🔵	io/questdb/client/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoop.java	51	71	71.83%
🔵	io/questdb/client/cutlass/qwp/client/sf/cursor/SegmentProvisioner.java	56	65	86.15%
🔵	io/questdb/client/cutlass/qwp/client/sf/cursor/SegmentManager.java	25	29	86.21%
🔵	io/questdb/client/cutlass/qwp/client/sf/cursor/CursorSendEngine.java	23	26	88.46%
🔵	io/questdb/client/cutlass/http/client/WebSocketClient.java	26	29	89.66%
🔵	io/questdb/client/cutlass/qwp/client/sf/cursor/SegmentRing.java	55	58	94.83%
🔵	io/questdb/client/cutlass/qwp/client/sf/cursor/AckWaiters.java	24	25	96.00%
🔵	io/questdb/client/cutlass/qwp/client/QwpWebSocketSender.java	55	56	98.21%
🔵	io/questdb/client/network/JavaTlsClientSocket.java	4	4	100.00%
🔵	io/questdb/client/cutlass/qwp/client/sf/cursor/SenderErrorDispatcher.java	1	1	100.00%
🔵	io/questdb/client/cutlass/qwp/client/sf/cursor/SenderConnectionDispatcher.java	1	1	100.00%
🔵	io/questdb/client/cutlass/qwp/client/sf/cursor/SenderProgressDispatcher.java	10	10	100.00%
🔵	io/questdb/client/network/Socket.java	1	1	100.00%