Pin catch-up id tiling and fix stale comments

The split catch-up tests counted frames but never read their bytes, so
a chunk walk shipping the right NUMBER of frames while overlapping or
skipping ids passed them all. That is the silent-NULL corruption
vector: the server null-pads a gapped delta, and a row referencing a
padded id lands a NULL symbol with no error anywhere. Mutating the
walk's start-id advance proves the gap -- the alignment suite stayed
17/17 green while only the 25-second end-to-end test noticed.

CatchUpCapturingClient now keeps each frame's bytes, joining the two
slices of a multipart send, and assertCatchUpReassembles folds them
through QwpWireTestUtils.accumulateDeltaDictionary -- the decoder the
end-to-end handler already uses -- then compares per id. Comparing
content rather than ranges catches overlap, gap and shift alike, and
each frame is checked against the advertised cap. QwpWireTestUtils and
that one method turn public for the sf.cursor package; the rest of the
class stays package-private.

Add three tests: a three-way split whose chunks must tile [0, n)
exactly; an empty-dictionary reconnect that must ship no catch-up
frame, using cap 0 so the guard rather than an empty walk holds the
count at zero; and a variable-width split, the only test that catches
a walk assuming a uniform entry stride.

Cover multi-byte UTF-8 on the persist path, which nothing exercised:
every symbol in these suites was ASCII, where a symbol's UTF-8 byte
length and its char count agree. Replacing Utf8s.utf8Bytes with
String.length leaves DeltaDictRecoveryTest 23/23 green while the new
round-trip test fails on the truncated tail. The literals carry a
runtime self-check so the test cannot decay into an ASCII one should
the file ever lose its encoding.

Drop a comment block left stranded above WebSocketResponse when its
field was removed, and rewrite two that described an entry-end index
this class never had. Correct the surviving Invariant B text: the
reconnect budget reaches connectWithRetry from ensureConnected's SYNC
branch and BackgroundDrainer's capability-gap deadline, never
buildAndConnect and never this loop.

Tear the torn-dict guard test down recursively. It builds the slot
layout under <dir>/default, which the flat helper cannot remove, so
every run left the whole tree behind.

Production behaviour is unchanged: every main/ hunk is comment text.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: glasstiger

core/src/main/java/io/questdb/client/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoop.java

@@ -249,11 +249,6 @@ public final class CursorWebSocketSendLoop implements QuietCloseable {
     private final ReconnectFactory reconnectFactory;
     private final long reconnectInitialBackoffMillis;
     private final long reconnectMaxBackoffMillis;
-    // Retained for constructor symmetry and passed in by callers, but NOT
-    // consulted by the background loop: Invariant B removed the wall-clock
-    // give-up from connectLoop. The budget still bounds the blocking (non-lazy)
-    // initial connect via QwpWebSocketSender -> connectWithRetry, which takes it
-    // as an explicit argument rather than reading this field.
     private final WebSocketResponse response = new WebSocketResponse();
     private final ResponseHandler responseHandler = new ResponseHandler();
     private final CountDownLatch shutdownLatch = new CountDownLatch(1);
@@ -761,10 +756,14 @@ public CursorWebSocketSendLoop(WebSocketClient client, CursorSendEngine engine,
         // re-registers its dictionary from id 0 as it replays, so seeding the mirror would buy
         // nothing and cost a catch-up frame on every connection -- full-dict slots must stay
         // catch-up-free.
-        // The prefix seed above may be borrowed, while the entry-end index is always
-        // loop-owned native memory. Build/extend both inside one cleanup boundary:
-        // any allocation failure propagates out of the constructor before a caller
-        // can own the half-built loop, so nothing else could release them.
+        // The prefix seed above may still be borrowed from PersistedSymbolDict.
+        // Extending it inside this cleanup boundary is what makes that safe:
+        // ensureSentDictCapacity copy-on-writes a borrowed prefix into loop-OWNED
+        // native memory, so from that point a failure would leak it -- and an
+        // allocation failure here propagates out of the constructor before any
+        // caller can own the half-built loop, leaving nothing else able to
+        // release it. releaseSentDictBytes() in the catch frees exactly that
+        // mirror, and only when this loop owns it.
         try {
             if (engine.recoveredMaxSymbolDeltaStart() > 0L) {
                 int baseline = sentDictCount;
@@ -792,10 +791,12 @@ public CursorWebSocketSendLoop(WebSocketClient client, CursorSendEngine engine,
                     }
                 }
             }
-            // The entry-ends index is NOT built here. sendDictCatchUp is its only
-            // reader and builds it on demand, so a recovered slot that drains without
-            // ever reconnecting -- the normal case -- never pays the O(n) walk nor
-            // retains the 4-bytes-per-symbol index.
+            // No per-entry offset index is derived here, and none is kept anywhere:
+            // the mirror carries its own framing ([len varint][utf8] per entry), so
+            // sendDictCatchUp -- its only reader -- recovers each entry's span with a
+            // running pointer as it walks. A recovered slot that drains without ever
+            // reconnecting, the normal case, therefore never pays that O(n) walk and
+            // carries no index memory for it.
         } catch (Throwable t) {
             releaseSentDictBytes();
             throw t;
@@ -1575,9 +1576,12 @@ private void connectLoop(Throwable initial, String phase, long paceFirstAttemptM
         // drainers. Foreground senders retry them from asynchronous startup onward
         // so a credential or cluster capability rotation cannot stop the producer. SF
         // exhaustion is surfaced to the PRODUCER as append backpressure, never
-        // here. reconnect_max_duration_millis is intentionally NOT consulted: it
-        // bounds only the blocking (non-lazy) initial connect in
-        // QwpWebSocketSender.buildAndConnect, never this background loop.
+        // here. reconnect_max_duration_millis is intentionally NOT consulted by
+        // THIS loop. Its holders pass it explicitly where it does apply: the
+        // blocking (non-lazy) initial connect hands it to connectWithRetry (see
+        // QwpWebSocketSender.ensureConnected, the SYNC branch), and
+        // BackgroundDrainer converts it into the durable-ack capability-gap
+        // budget. Neither bounds this loop's steady-state reconnect.
         long backoffMillis = reconnectInitialBackoffMillis;
         if (paceFirstAttemptMillis > 0 && running) {
             // NACK-initiated recycle against a reachable server: pace the

core/src/test/java/io/questdb/client/test/cutlass/qwp/client/QwpWireTestUtils.java

@@ -32,9 +32,18 @@
 import static io.questdb.client.cutlass.qwp.protocol.QwpConstants.FLAG_DELTA_SYMBOL_DICT;
 import static io.questdb.client.cutlass.qwp.protocol.QwpConstants.HEADER_SIZE;
 
-final class QwpWireTestUtils {
+public final class QwpWireTestUtils {
 
-    static void accumulateDeltaDictionary(byte[] frame, List<String> dictionary) {
+    /**
+     * Folds one frame's symbol-dict delta into {@code dictionary}, exactly as the
+     * server does -- including padding a gap with nulls rather than rejecting it,
+     * so a caller can observe a gap instead of having it hidden. Public so the
+     * store-and-forward send-loop tests in the {@code sf.cursor} package can
+     * reassemble captured catch-up frames through the SAME decoder the
+     * end-to-end tests' server handler uses, rather than hand-rolling a second
+     * one that could drift from it.
+     */
+    public static void accumulateDeltaDictionary(byte[] frame, List<String> dictionary) {
         if (!hasDelta(frame)) {
             return;
         }

core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoopCatchUpAlignmentTest.java

@@ -34,6 +34,7 @@
 import io.questdb.client.network.PlainSocketFactory;
 import io.questdb.client.std.MemoryTag;
 import io.questdb.client.std.Unsafe;
+import io.questdb.client.test.cutlass.qwp.client.QwpWireTestUtils;
 import io.questdb.client.test.tools.TestUtils;
 import org.junit.After;
 import org.junit.Before;
@@ -166,25 +167,31 @@ public void testSplitCatchUpStagesOnlyPrefixAcrossReconnects() throws Exception
             try (CursorSendEngine engine = newEngine()) {
                 CursorWebSocketSendLoop loop = newLoop(engine, client);
                 try {
-                    seedMirror(loop, TestUtils.repeat("x", 3_000), TestUtils.repeat("y", 3_000));
+                    String symX = TestUtils.repeat("x", 3_000);
+                    String symY = TestUtils.repeat("y", 3_000);
+                    seedMirror(loop, symX, symY);
 
                     invokeSetWireBaselineWithCatchUp(loop, 0L);
                     assertEquals("the small cap must split the dictionary", 2, client.framesSent);
                     assertEquals("catch-up must send the symbol bytes as a second payload slice",
                             2, client.multipartFramesSent);
                     assertEquals("the split chunks need one small prefix buffer",
                             1, loop.catchUpFrameGrowthCount());
+                    // Splitting is only correct if the chunks reassemble gap-free.
+                    assertCatchUpReassembles(client, symX, symY);
 
                     client.cap = 7_000;
                     invokeSetWireBaselineWithCatchUp(loop, 0L);
                     assertEquals("the larger cap must combine the dictionary", 3, client.framesSent);
                     assertEquals("combining symbols must not grow the prefix-only buffer",
                             1, loop.catchUpFrameGrowthCount());
+                    assertCatchUpReassembles(client, symX, symY);
 
                     invokeSetWireBaselineWithCatchUp(loop, 0L);
                     assertEquals("the next reconnect sends one combined frame", 4, client.framesSent);
                     assertEquals("the prefix buffer must be reused across reconnects",
                             1, loop.catchUpFrameGrowthCount());
+                    assertCatchUpReassembles(client, symX, symY);
                 } finally {
                     // assertMemoryLeak verifies that close releases the retained buffer.
                     loop.close();
@@ -193,6 +200,113 @@ public void testSplitCatchUpStagesOnlyPrefixAcrossReconnects() throws Exception
         });
     }
 
+    @Test
+    public void testSplitCatchUpChunksTileTheDictionaryExactly() throws Exception {
+        // Three chunks rather than two: a start-id error that happens to cancel
+        // out across a single boundary cannot survive two of them, and only a
+        // multi-chunk walk exercises the accumulating "chunkStartId +=
+        // chunkSymbols" drift. Six 100-byte symbols under a cap that fits two
+        // per frame; the reassembly must return all six, in order, with no null.
+        TestUtils.assertMemoryLeak(() -> {
+            String[] symbols = new String[6];
+            for (int i = 0; i < symbols.length; i++) {
+                symbols[i] = TestUtils.repeat(String.valueOf((char) ('a' + i)), 100);
+            }
+            CatchUpCapturingClient client = new CatchUpCapturingClient(250);
+            try (CursorSendEngine engine = newEngine()) {
+                CursorWebSocketSendLoop loop = newLoop(engine, client);
+                try {
+                    seedMirror(loop, symbols);
+                    invokeSetWireBaselineWithCatchUp(loop, 0L);
+                    assertEquals("the cap must force a three-way split", 3, client.framesSent);
+                    assertCatchUpReassembles(client, symbols);
+                } finally {
+                    loop.close();
+                }
+            }
+        });
+    }
+
+    @Test
+    public void testEmptyDictionaryReconnectSendsNoCatchUpFrame() throws Exception {
+        // The sentDictCount > 0 half of setWireBaselineWithCatchUp's gate. Every
+        // other test here seeds at least one symbol, so the empty-dictionary
+        // branch -- a delta-enabled connection that has not registered anything
+        // yet, i.e. every reconnect before the first symbol -- was never
+        // exercised. Emitting a zero-entry catch-up frame there would burn a wire
+        // sequence and, via fsnAtZero = replayStart - catchUpFrames, shift the
+        // baseline so the first real frame no longer lands on replayStart.
+        TestUtils.assertMemoryLeak(() -> {
+            // cap 0 ("server advertises no cap") on purpose. Under a positive cap
+            // sendDictCatchUp would walk zero entries and return 0 frames anyway,
+            // so both assertions below would hold even with the sentDictCount > 0
+            // conjunct deleted -- the test would pin nothing. Cap 0 takes the
+            // fast path that sends one whole-dictionary frame unconditionally, so
+            // only the guard keeps the frame count at zero.
+            CatchUpCapturingClient client = new CatchUpCapturingClient(0);
+            try (CursorSendEngine engine = newEngine()) {
+                CursorWebSocketSendLoop loop = newLoop(engine, client);
+                try {
+                    // Deliberately no seedMirror: sentDictCount stays 0.
+                    invokeSetWireBaselineWithCatchUp(loop, 7L);
+                    assertEquals("an empty dictionary must not ship a catch-up frame",
+                            0, client.framesSent);
+                    assertEquals("the baseline must stay at replayStart when nothing is re-registered",
+                            7L, loop.fsnAtZero());
+                } finally {
+                    loop.close();
+                }
+            }
+        });
+    }
+
+    @Test
+    public void testCatchUpSplitsVariableWidthEntriesWithoutDrift() throws Exception {
+        // Every other split test here uses uniformly-sized symbols, so the chunk
+        // walk always advances by the same stride and a span miscalculation could
+        // cancel out. These entries are 14, 7, 11 and 22 bytes, so each hop is a
+        // different width and the walk has to resume mid-dictionary at an
+        // irregular boundary; the reassembly then pins the result per id.
+        //
+        // The symbols are multi-byte UTF-8 because that is a convenient source of
+        // widths that differ from their char counts, and it exercises the wire
+        // framing end to end -- NOT because sendDictCatchUp could confuse bytes
+        // for chars: that method holds no String, char or length() at all, only
+        // pointer arithmetic over [len varint][utf8]. The byte-vs-char hazard
+        // lives on the persist path and is covered by
+        // PersistedSymbolDictTest.testMultiByteUtf8SymbolsRoundTripAcrossReopen.
+        TestUtils.assertMemoryLeak(() -> {
+            String[] symbols = {
+                    // entry width below = 1-byte length varint + the UTF-8 bytes
+                    "températures",           // 12 chars, 13 bytes -> 14
+                    "東京",                    //  2 chars,  6 bytes ->  7
+                    "sensor🔥",              //  7 chars, 10 bytes -> 11
+                    "ascii_after_multibyte"   // 21 chars, 21 bytes -> 22; a drift shows up here
+            };
+            // Self-check: prove the literals really are multi-byte at runtime, so
+            // the widths stay irregular even if the source file ever loses its
+            // encoding and they collapse to single-byte '?'.
+            for (String s : symbols) {
+                if (!"ascii_after_multibyte".equals(s)) {
+                    assertTrue("expected a multi-byte UTF-8 symbol, got pure ASCII: " + s,
+                            s.getBytes(StandardCharsets.UTF_8).length > s.length());
+                }
+            }
+            CatchUpCapturingClient client = new CatchUpCapturingClient(80);
+            try (CursorSendEngine engine = newEngine()) {
+                CursorWebSocketSendLoop loop = newLoop(engine, client);
+                try {
+                    seedMirror(loop, symbols);
+                    invokeSetWireBaselineWithCatchUp(loop, 0L);
+                    assertTrue("the cap must split the dictionary", client.framesSent > 1);
+                    assertCatchUpReassembles(client, symbols);
+                } finally {
+                    loop.close();
+                }
+            }
+        });
+    }
+
     @Test
     public void testTransientCatchUpSendFailureIsRetriableNotTerminal() throws Exception {
         // A transient wire failure WHILE shipping the catch-up (the fresh
@@ -931,6 +1045,43 @@ private CursorSendEngine newEngine() {
         return new CursorSendEngine(tmpDir, 16_384);
     }
 
+    /**
+     * Reassembles the frames captured since the last call exactly as the server
+     * would -- through the same {@link QwpWireTestUtils#accumulateDeltaDictionary}
+     * the end-to-end tests' handler uses -- and asserts the result is the seeded
+     * dictionary, dense and in order.
+     * <p>
+     * This is what frame counting cannot do. A catch-up split ships its chunks as
+     * {@code [deltaStart, deltaStart+count)} ranges that must tile {@code [0, n)}
+     * exactly; an off-by-one in the walk's start id keeps the frame COUNT intact
+     * while overlapping a range (an id silently takes its neighbour's symbol) or
+     * skipping one (the server null-pads it, and rows referencing it land a NULL
+     * symbol value). Comparing the reassembled dictionary catches all three
+     * shapes -- overlap, gap and shift -- because it compares content per id, not
+     * just the ranges.
+     */
+    private static void assertCatchUpReassembles(CatchUpCapturingClient client, String... expected) {
+        List<String> rebuilt = new ArrayList<>();
+        for (byte[] frame : client.capturedFrames) {
+            // Tiling correctly is necessary but not sufficient: the split exists to
+            // keep every frame under the server's advertised batch cap, and a
+            // perfectly contiguous split that ignores the cap would otherwise only
+            // be caught indirectly, by a frame-count assertion.
+            if (client.cap > 0) {
+                assertTrue("catch-up frame of " + frame.length
+                        + " bytes exceeds the advertised cap " + client.cap,
+                        frame.length <= client.cap);
+            }
+            QwpWireTestUtils.accumulateDeltaDictionary(frame, rebuilt);
+        }
+        client.capturedFrames.clear();
+        assertEquals("reassembled dictionary size", expected.length, rebuilt.size());
+        for (int i = 0; i < expected.length; i++) {
+            assertEquals("symbol at id " + i + " (a null here is a gap the server would"
+                    + " turn into a NULL symbol value)", expected[i], rebuilt.get(i));
+        }
+    }
+
     // Populates the loop's native sent-dictionary mirror with {@code symbols} in
     // the on-wire [len varint][utf8] layout, so setWireBaselineWithCatchUp sees a
     // non-empty dictionary to re-register. loop.close() frees it.
@@ -980,6 +1131,7 @@ private static final class CatchUpCapturingClient extends WebSocketClient {
         // Mutable so a test can model a rolling-cap cluster: raise it for a node that
         // accepts the dictionary, lower it for a smaller-cap node that cap-gaps.
         private int cap;
+        private final List<byte[]> capturedFrames = new ArrayList<>();
         private final Runnable onCapRead;
         private final boolean throwOnSend;
         private int framesSent;
@@ -1016,6 +1168,7 @@ public int getServerQwpVersion() {
         @Override
         public void sendBinary(long dataPtr, int length) {
             recordSend();
+            capture(dataPtr, length, 0L, 0);
         }
 
         @Override
@@ -1027,6 +1180,26 @@ public void sendBinary(
         ) {
             multipartFramesSent++;
             recordSend();
+            capture(firstPtr, firstLength, secondPtr, secondLength);
+        }
+
+        /**
+         * Keeps the bytes of every frame sent, joining the two slices of a
+         * multipart send back into the single frame the server would see.
+         * Counting frames alone cannot detect the failure that matters here: an
+         * off-by-one in the chunk walk's start id ships the same NUMBER of
+         * frames while overlapping or skipping ids, and the server null-pads a
+         * skipped id into a silent NULL symbol.
+         */
+        private void capture(long firstPtr, int firstLength, long secondPtr, int secondLength) {
+            byte[] frame = new byte[firstLength + secondLength];
+            for (int i = 0; i < firstLength; i++) {
+                frame[i] = Unsafe.getUnsafe().getByte(firstPtr + i);
+            }
+            for (int i = 0; i < secondLength; i++) {
+                frame[firstLength + i] = Unsafe.getUnsafe().getByte(secondPtr + i);
+            }
+            capturedFrames.add(frame);
         }
 
         private void recordSend() {

core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoopTornDictGuardTest.java

@@ -72,7 +72,11 @@ public void setUp() {
 
     @After
     public void tearDown() {
-        TestUtils.removeTmpDir(sfDir);
+        // Recursive: this test builds the store-and-forward slot layout
+        // (<dir>/default/... plus <dir>/.slot-locks/...), and the flat variant
+        // cannot remove a non-empty subdirectory -- it also discards its result,
+        // so the whole tree survived every run unnoticed.
+        TestUtils.removeTmpDirRec(sfDir);
     }
 
     @Test