Cover split-preflight and torn-dict guard gaps

Address review findings on the delta symbol-dictionary change.

Tests (close two coverage gaps):
- SelfSufficientFramesTest pins the split-flush pre-flight baseline
  advance: a large-delta two-table split whose second frame fits only
  with an empty delta. Removing the simBaseline advance mis-sizes that
  frame with the delta and wrongly rejects a shippable batch.
- CursorWebSocketSendLoopOrphanTailTest pins that a zero-count delta
  frame anchors recoveredMaxSymbolId at its baseline, not 0. Skipping
  it would under-strand a torn dictionary and risk a silent id shift.

Docs (correct load-bearing recovery semantics):
- seedGlobalDictionaryFromPersisted no longer claims the drainer always
  self-heals: it does so only when a surviving frame straddles the tear;
  a higher-baseline survivor quarantines the slot, a deliberate
  conservative over-strand.
- maxSymbolDeltaEnd documents that a zero-count delta frame contributes
  its deltaStart, not 0.
- The catch-up cap-gap budget documents that a transient reconnect never
  increments or resets the counter (only a successful catch-up resets
  it), so a transient cannot burn the terminal budget.

Hardening:
- writeVarint and putVarint assert a non-negative value: the signed loop
  writes one truncated byte for a negative long while varintSize returns
  10.
- PersistedSymbolDict.open diverts a file at exactly Integer.MAX_VALUE to
  a clean re-create instead of a doomed 2GB malloc (> becomes >=).
- readLoadedSymbols bounds its varint decode (shift > 35) like its
  sibling parsers.

Production behavior changes are limited to the two varint asserts, the
>= boundary, and the defensive varint bound; the rest is tests and
documentation. Full delta-dict suite green on JDK 25 with -ea.

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

Author: glasstiger

core/src/main/java/io/questdb/client/cutlass/qwp/client/NativeBufferWriter.java

@@ -81,8 +81,14 @@ public static int varintSize(long value) {
      * {@code addr} and returns the address just past the last byte. The canonical
      * raw-address varint writer shared by the SF cursor's persisted dictionary and
      * catch-up frame builder.
+     * <p>
+     * {@code value} must be non-negative: the signed {@code value > 0x7F} loop emits
+     * a SINGLE truncated byte for a negative long, whereas {@link #varintSize}
+     * returns 10 for it -- a size/write mismatch that would corrupt the stream. All
+     * callers pass ids/lengths/counts (non-negative); the assert pins that contract.
      */
     public static long writeVarint(long addr, long value) {
+        assert value >= 0 : "unsigned LEB128 varint requires a non-negative value: " + value;
         while (value > 0x7F) {
             Unsafe.getUnsafe().putByte(addr++, (byte) ((value & 0x7F) | 0x80));
             value >>>= 7;
@@ -320,6 +326,7 @@ public void putUtf8(CharSequence value) {
      */
     @Override
     public void putVarint(long value) {
+        assert value >= 0 : "unsigned LEB128 varint requires a non-negative value: " + value;
         ensureCapacity(10); // max varint bytes
         long addr = bufferPtr + position;
         while (value > 0x7F) {

core/src/main/java/io/questdb/client/cutlass/qwp/client/QwpWebSocketSender.java

@@ -3876,9 +3876,23 @@ private void resetSymbolDictStateForNewConnection() {
      * mirror, leaving only this producer diverged. Detect it here -- the surviving
      * frames reference an id at or beyond the recovered dictionary size -- and fail
      * clean: the affected data must be resent, matching the design's torn-dict
-     * "resend required" contract. The recovered data is not lost; the background
-     * drainer still drains this slot (its mirror self-heals from the frames). Only a
-     * host crash reaches this -- a process crash keeps the page cache, so the
+     * "resend required" contract.
+     * <p>
+     * The background drainer self-heals the mirror ONLY when a surviving frame
+     * STRADDLES the tear ({@code deltaStart <= pd.size() < deltaStart + deltaCount}):
+     * such a frame carries the torn-off ids in its own delta and
+     * {@code accumulateSentDict} re-registers them, so the drainer drains the slot.
+     * But when the symbol-introducing frames were already acked and trimmed and only
+     * a HIGHER-baseline frame survives ({@code deltaStart > pd.size()} -- e.g. a
+     * commit or a symbol-reusing frame, since {@code beginMessage} always sets the
+     * delta flag), the drainer's own replay guard ({@code deltaStart > sentDictCount})
+     * fires too and quarantines the slot: the recorded bytes are not silently lost,
+     * but the slot is NOT auto-drained -- it must be resent. That is a deliberate
+     * CONSERVATIVE over-strand -- the guard keys on {@code deltaStart}, not on the
+     * frame's actual highest referenced id, to avoid parsing row data at recovery,
+     * so it may reject a frame whose rows reference only ids the truncated
+     * dictionary still holds. It fails clean rather than risk a silent id shift.
+     * Only a host crash reaches this -- a process crash keeps the page cache, so the
      * write-ahead ordering keeps the dictionary a superset of the frames.
      */
     private void seedGlobalDictionaryFromPersisted(PersistedSymbolDict pd) {

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

@@ -133,17 +133,30 @@ public final class CursorWebSocketSendLoop implements QuietCloseable {
      */
     public static final long DEFAULT_POISON_MIN_ESCALATION_WINDOW_MILLIS = 5_000L;
     private static final Logger LOG = LoggerFactory.getLogger(CursorWebSocketSendLoop.class);
-    // Settle budget for the symbol-dict catch-up cap gap: how many CONSECUTIVE
-    // reconnect attempts may find a single dictionary entry too large for the fresh
-    // server's advertised batch cap before the sender latches a terminal. A
-    // homogeneous cluster never trips it -- an entry that fit its data frame under a
-    // cap always fits its bare catch-up frame under that same cap -- so this only
-    // affects a heterogeneous / rolling-cap cluster, where a failover to a
-    // smaller-cap node can hit it for an entry an earlier node accepted. Retrying
-    // rides out the transient window until a larger-cap node returns; only a
-    // persistent gap (every reachable node too small for this many attempts) latches
-    // terminal, matching the orphan drainer's DEFAULT_MAX_DURABLE_ACK_MISMATCH_ATTEMPTS.
-    // A successful catch-up resets the counter (see sendDictCatchUp).
+    // Settle budget for the symbol-dict catch-up cap gap: how many cap-gap attempts
+    // -- catch-ups that reached a fresh server and found a single dictionary entry
+    // too large for its advertised batch cap -- may occur, with no intervening
+    // SUCCESSFUL catch-up, before the sender latches a terminal. This is a SANCTIONED
+    // terminal (a genuine cluster batch-size capability gap), the connect-time analog
+    // of the orphan drainer's durable-ack capability gap
+    // (DEFAULT_MAX_DURABLE_ACK_MISMATCH_ATTEMPTS). A homogeneous cluster never trips
+    // it -- an entry that fit its data frame under a cap always fits its bare catch-up
+    // frame under that same cap -- so it only affects a heterogeneous / rolling-cap
+    // cluster, where a failover to a smaller-cap node can hit it for an entry an
+    // earlier node accepted. Retrying rides out the transient window until a
+    // larger-cap node returns; only a persistent gap (this many cap gaps with no
+    // successful catch-up in between) latches.
+    //
+    // Budget accounting (satisfies "a transient must never burn the terminal
+    // budget"): catchUpCapGapAttempts increments ONLY inside sendDictCatchUp when a
+    // node is reached and an entry is oversized, and resets ONLY when a catch-up fully
+    // succeeds. A TRANSIENT reconnect (connect refuse, upgrade/role failure) never
+    // reaches the catch-up, so it NEITHER increments nor resets the counter -- it only
+    // lengthens the wall-clock settle window. The terminal therefore always requires
+    // this many GENUINE cap gaps; a transient can never inflate it. Deliberately NOT
+    // reset on a mere successful RECONNECT: a reconnect to the small-cap node itself
+    // produces the cap gap, so resetting there would stop a persistent gap from ever
+    // latching -- the reset must gate on a successful CATCH-UP, not on connecting.
     private static final int MAX_CATCHUP_CAP_GAP_ATTEMPTS = 16;
     // Hard ceiling for the lifetime-monotonic sent-dictionary mirror. The mirror
     // fields are int, so it cannot exceed Integer.MAX_VALUE bytes; reaching even
@@ -218,11 +231,12 @@ public final class CursorWebSocketSendLoop implements QuietCloseable {
     // for the connection's lifetime (a reconnect may need the whole dictionary at
     // any moment), so it cannot be dropped; it is an intentional cost of the feature.
     private final boolean deltaDictEnabled;
-    // Consecutive reconnect attempts whose symbol-dict catch-up found an entry too
-    // large for the fresh server's batch cap (see MAX_CATCHUP_CAP_GAP_ATTEMPTS). A
-    // successful catch-up resets it; it is NOT reset per connection -- it measures
-    // the cap-gap episode across reconnects so a persistent gap eventually latches.
-    // I/O-thread-only.
+    // Cap-gap attempts -- catch-ups that reached a node and found an entry too large
+    // for its batch cap -- since the last SUCCESSFUL catch-up (see
+    // MAX_CATCHUP_CAP_GAP_ATTEMPTS for the full budget accounting). A successful
+    // catch-up resets it (sendDictCatchUp); a transient reconnect neither increments
+    // nor resets it. NOT reset per connection -- it measures the cap-gap episode
+    // across reconnects so a persistent gap eventually latches. I/O-thread-only.
     private int catchUpCapGapAttempts;
     // True once a real ring frame (data or commit) has been sent on the CURRENT
     // connection, as opposed to only the dictionary catch-up. The catch-up

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

@@ -523,9 +523,16 @@ public long findLastFrameFsnWithoutPayloadFlag(int flagsOffset, int flagMask, in
     }
 
     /**
-     * Highest {@code deltaStart + deltaCount} (one past the highest symbol id) that
-     * any symbol-dict delta frame in this segment references, or {@code 0} when no
-     * such frame carries a symbol. Read-only walk over the recovered frames, used
+     * Highest {@code deltaStart + deltaCount} (one past the highest symbol id) any
+     * delta-flagged frame in this segment carries, or {@code 0} only when NO
+     * delta-flagged frame is present. A frame with {@code deltaCount == 0} (a commit
+     * or a symbol-reusing frame -- the encoder always sets the delta flag) still
+     * contributes its {@code deltaStart}, i.e. the producer's baseline at encode
+     * time, NOT 0. That is deliberate: it anchors the torn-dict guard at that
+     * baseline so a dictionary torn below it is detected -- a conservative
+     * over-strand (it may over-reject a frame whose rows reference only surviving
+     * ids), never an under-strand that could silently shift the dense id map.
+     * Read-only walk over the recovered frames, used
      * once at recovery to detect a persisted {@code .symbol-dict} torn (host crash,
      * out-of-order page loss) below the ids the surviving frames still reference: if
      * the max here reaches at or beyond the recovered dictionary size, a resuming

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

@@ -170,17 +170,19 @@ public static PersistedSymbolDict open(String slotDir) {
     public static PersistedSymbolDict open(FilesFacade ff, String slotDir) {
         String filePath = slotDir + "/" + FILE_NAME;
         long existing = ff.exists(filePath) ? ff.length(filePath) : -1L;
-        // A dictionary that legitimately grew past Integer.MAX_VALUE cannot be
-        // reopened: openExisting reads it into ONE int-sized native buffer, and
-        // the (int) cast of a >2GB length is either negative (malloc rejects it),
-        // exactly zero (getInt then reads 4 bytes past a zero-size allocation), or
-        // a small positive prefix (whose validLen < len branch would then
-        // DESTRUCTIVELY truncate the multi-GB file). Recreate empty instead --
-        // fail-clean, exactly like every other unreadable-file case here, so the
-        // sender falls back to full self-sufficient frames. Reaching this needs
-        // ~100M+ distinct symbols on one slot (far past realistic symbol
-        // cardinality); the guard keeps the read/write size boundary safe anyway.
-        if (existing > Integer.MAX_VALUE) {
+        // A dictionary that grew to or past Integer.MAX_VALUE cannot be reopened:
+        // openExisting reads it into ONE int-sized native buffer. PAST 2GiB the
+        // (int) cast is either negative (malloc rejects it), exactly zero (getInt
+        // then reads 4 bytes past a zero-size allocation), or a small positive prefix
+        // (whose validLen < len branch would then DESTRUCTIVELY truncate the multi-GB
+        // file); AT exactly Integer.MAX_VALUE the cast is exact but the ~2GB malloc is
+        // doomed to OutOfMemoryError. The >= guard short-circuits every case to a
+        // clean re-create instead of the doomed allocation -- fail-clean, exactly like
+        // every other unreadable-file case here, so the sender falls back to full
+        // self-sufficient frames. Reaching this needs ~100M+ distinct symbols on one
+        // slot (far past realistic symbol cardinality); the guard keeps the read/write
+        // size boundary safe anyway.
+        if (existing >= Integer.MAX_VALUE) {
             LOG.warn("symbol dict {} too large ({} bytes) to reopen; recreating empty", filePath, existing);
             return openFresh(ff, filePath);
         }
@@ -463,6 +465,13 @@ public ObjList<String> readLoadedSymbols() {
                     break;
                 }
                 shift += 7;
+                if (shift > 35) {
+                    // Bound the varint like decodeVarint / appendRawEntries /
+                    // readVarintAt: a canonical length is <= 5 bytes. open() already
+                    // CRC-validated these bytes, so this is defensive only; the
+                    // p + len > limit check below then rejects the over-long run.
+                    break;
+                }
             }
             if (p + len > limit) {
                 break; // defensive: torn tail (should not happen past parse in open)

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

@@ -381,6 +381,75 @@ public void testFileModeSplitPersistsDictBeforePublish() throws Exception {
         }
     }
 
+    @Test
+    public void testSplitPreflightAdvancesBaselineSoLaterFramesArentSizedWithTheDelta() throws Exception {
+        // Regression for the split pre-flight baseline advance in flushPendingRowsSplit
+        // (the "Mirror advanceSentMaxSymbolId" step). Only the FIRST split frame ships
+        // the batch's symbol-dict delta; the rest ship an EMPTY delta and reference ids
+        // the first frame already registered. The pre-flight size pass must advance
+        // simBaseline after the first table so it STOPS adding combinedDeltaEntriesLen
+        // to the later frames' estimated sizes. Without that advance, a later table
+        // whose real (empty-delta) frame fits the cap is mis-estimated as still carrying
+        // the whole delta and wrongly rejected with "single table batch too large" --
+        // discarding a legitimately shippable batch (fail-closed data loss).
+        //
+        // Shape (memory mode, delta enabled): a LARGE combined delta (two ~64-char
+        // symbols) rides only the first split frame. The first table (added first, so
+        // the first split frame) has a tiny body, so delta + body1 fits the cap. The
+        // second table has a big body: body2 alone fits the cap, but delta + body2 does
+        // NOT. The real code splits and ships both frames; the un-advanced pre-flight
+        // would size the second frame WITH the delta and throw. Table order is
+        // insertion order (CharSequenceObjHashMap.keys()), so t1 is the delta frame.
+        assertMemoryLeak(() -> {
+            CapturingHandler handler = new CapturingHandler();
+            try (TestWebSocketServer server = new TestWebSocketServer(handler)) {
+                server.setAdvertisedMaxBatchSize(200);
+                int port = server.getPort();
+                server.start();
+                Assert.assertTrue(server.awaitStart(5, TimeUnit.SECONDS));
+
+                // Two long symbols => a large combined delta section (~130 bytes) that
+                // rides ONLY the first split frame. The symbol STRINGS live in the delta,
+                // not in either table body (the body carries only the varint global id).
+                String longSymA = new String(new char[64]).replace('\0', 'a');
+                String longSymB = new String(new char[64]).replace('\0', 'b');
+                String bigPad = new String(new char[100]).replace('\0', 'x');
+                // auto_flush off so both rows batch into one flush (byte-based auto-flush
+                // is otherwise clamped under the cap and would flush before the split).
+                try (Sender sender = Sender.fromConfig("ws::addr=localhost:" + port
+                        + ";auto_flush_bytes=off;auto_flush_rows=1000000;auto_flush_interval=60000;")) {
+                    // t1 (added first -> first split frame): carries the whole delta but a
+                    // tiny body, so delta + body1 fits the 200-byte cap.
+                    sender.table("t1").symbol("s", longSymA).longColumn("v", 1L).atNow();
+                    // t2 (second split frame): empty delta but a big body. body2 alone
+                    // fits the cap; delta + body2 does NOT -- the mis-size the advance
+                    // prevents.
+                    sender.table("t2").symbol("s", longSymB).stringColumn("p", bigPad).longColumn("v", 2L).atNow();
+                    // Must NOT throw: with the baseline advanced, t2's frame is sized
+                    // WITHOUT the delta and fits. A broken advance throws "too large" here.
+                    sender.flush();
+                    waitFor(() -> handler.batches.size() >= 2, 5_000);
+                }
+
+                Assert.assertEquals("the batch must split into 2 frames, neither spuriously rejected",
+                        2, handler.batches.size());
+                byte[] f1 = handler.batches.get(0);
+                byte[] f2 = handler.batches.get(1);
+                // First split frame ships the whole delta (both new symbols, ids 0 and 1).
+                Assert.assertEquals("first split frame deltaStart must be 0",
+                        0, readVarint(f1, DELTA_START_OFFSET));
+                Assert.assertEquals("first split frame ships both new symbols",
+                        2, readVarint(f1, DELTA_START_OFFSET + 1));
+                // Second split frame carries an EMPTY delta above the advanced baseline --
+                // the whole point: it is not re-sized (or re-sent) with the delta.
+                Assert.assertEquals("second split frame deltaStart must be 2 (baseline advanced)",
+                        2, readVarint(f2, DELTA_START_OFFSET));
+                Assert.assertEquals("second split frame carries no new symbols",
+                        0, readVarint(f2, DELTA_START_OFFSET + 1));
+            }
+        });
+    }
+
     private static int readVarint(byte[] buf, int offset) {
         // Simple unsigned varint decode — sufficient for small values.
         int result = 0;