Recover full-dict SF slots without bricking build

A store-and-forward slot written in full-dict fallback -- the
.symbol-dict could not open when writing, so every frame re-ships the
whole dictionary from id 0 -- leaves self-sufficient frames behind but
no side-file. On recovery the engine opens a fresh empty .symbol-dict,
so the surviving frames out-reach it, and the sender's seed-time guard
mistook the empty dictionary for a host-crash tear: it failed
Sender.build() for a slot the orphan drainer drains fine.

CursorSendEngine now tells the two cases apart with a new
maxSymbolDeltaStart walk (MmapSegment, SegmentRing). When the persisted
dictionary is a subset of the ids the frames reference but every such
frame is self-sufficient (maxSymbolDeltaStart == 0), the engine discards
the empty side-file so isDeltaDictEnabled() reports false and the slot
recovers in full-dict mode, exactly how it was written. A genuine torn
delta dictionary keeps a deltaStart > 0 frame and still fails clean.

testFullDictFramesRecoverInFullDictModeInsteadOfBricking covers the new
path -- red without the discard, green with it. The torn-delta
fail-clean tests are unchanged.

Also regroups CursorWebSocketSendLoop's six new mutable delta-dict
fields with the other mutable fields (deltaDictEnabled stays final, in
its alphabetical slot) and adds the missing thousands separator to a
test's 16_384 batch-size literal.

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

Author: glasstiger

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

@@ -3900,14 +3900,23 @@ private void seedGlobalDictionaryFromPersisted(PersistedSymbolDict pd) {
             return;
         }
         // Run the torn-dictionary guard BEFORE the empty-dictionary short-circuit
-        // below: a TOTAL tear (pd.size() == 0) with surviving symbol-bearing frames
-        // must fail clean too, not slip through. Such a frame starts at deltaStart=0
-        // and self-heals the I/O-thread catch-up mirror, so the send loop's replay
-        // guard (deltaStart > sentDictCount) never fires -- this seed-time guard is
-        // then the only defense against the producer resuming unseeded and silently
-        // reusing ids the frames already define. A genuinely empty slot (no
-        // symbol-bearing frames) has recoveredMaxSymbolId() == -1, so -1 >= 0 is false
-        // and the pd.size() == 0 return below still fires.
+        // below: a TOTAL tear (pd.size() == 0) of a DELTA dictionary with surviving
+        // symbol-bearing frames must fail clean too, not slip through. Such a frame
+        // starts at deltaStart=0 and self-heals the I/O-thread catch-up mirror, so the
+        // send loop's replay guard (deltaStart > sentDictCount) never fires -- this
+        // seed-time guard is then the only defense against the producer resuming
+        // unseeded and silently reusing ids the frames already define. A genuinely
+        // empty slot (no symbol-bearing frames) has recoveredMaxSymbolId() == -1, so
+        // -1 >= 0 is false and the pd.size() == 0 return below still fires.
+        //
+        // This guard fires ONLY for a torn DELTA dictionary. The other way the frames
+        // can out-reach the recovered dictionary -- a slot written in FULL-DICT
+        // fallback (the dictionary never opened when writing, every frame
+        // self-sufficient at deltaStart=0), then recovered against a fresh empty one --
+        // is caught upstream in CursorSendEngine, which discards the empty side-file so
+        // isDeltaDictEnabled() is false and this seed never runs. Those frames need no
+        // dictionary; failing clean here would needlessly brick build() for a slot the
+        // orphan drainer drains fine.
         if (cursorEngine != null && cursorEngine.recoveredMaxSymbolId() >= pd.size()) {
             throw new LineSenderException(
                     "recovered store-and-forward symbol dictionary is a subset of the surviving frames "

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

@@ -338,6 +338,36 @@ private CursorSendEngine(String sfDir, long segmentSizeBytes, SegmentManager man
                         QwpConstants.FLAG_DELTA_SYMBOL_DICT,
                         QwpConstants.HEADER_SIZE,
                         recoveredCommitBoundaryFsn) - 1L;
+                // Full-dict-fallback recovery. When the persisted .symbol-dict is a
+                // SUBSET of the ids the surviving frames reference
+                // (recoveredMaxSymbolId >= its size) YET every such frame is
+                // self-sufficient (maxSymbolDeltaStart == 0 -- a full-dict frame that
+                // re-registers its dictionary from id 0), the slot was written in
+                // full-dict fallback: the dictionary never opened when writing, so no
+                // side-file exists and this recovery opened a FRESH EMPTY one. Those
+                // frames replay with no dictionary, so discard the empty side-file and
+                // recover in full-dict mode -- isDeltaDictEnabled() then reports false
+                // and the producer + send loop both run full-dict, exactly as the slot
+                // was written. Without this the sender's seed-time guard would treat the
+                // empty dictionary as a host-crash tear and brick build(), even though
+                // the orphan drainer drains the same frames fine. A genuine torn DELTA
+                // dictionary keeps a frame with deltaStart > 0 (maxSymbolDeltaStart > 0)
+                // and is NOT discarded here: it still fails clean at seed time, since
+                // the ids its delta frames reference cannot be rebuilt without the lost
+                // dictionary. The recoveredMaxSymbolId >= size guard means this never
+                // fires for a slot whose dictionary is intact, nor for an empty slot
+                // (recoveredMaxSymbolId == -1). Single-threaded; before the I/O loop.
+                if (persistedDictInProgress != null
+                        && recoveredMaxSymbolId >= persistedDictInProgress.size()
+                        && recovered.maxSymbolDeltaStart(
+                        QwpConstants.MAGIC_MESSAGE,
+                        QwpConstants.HEADER_OFFSET_FLAGS,
+                        QwpConstants.FLAG_DELTA_SYMBOL_DICT,
+                        QwpConstants.HEADER_SIZE,
+                        recoveredCommitBoundaryFsn) == 0L) {
+                    persistedDictInProgress.close();
+                    persistedDictInProgress = null;
+                }
             } else {
                 // Fresh start with no recovered segments. Any stale
                 // watermark from a prior fully-drained session refers

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

@@ -168,6 +168,11 @@ public final class CursorWebSocketSendLoop implements QuietCloseable {
      * Throttle "reconnect attempt N failed" WARN logs to one per 5 s.
      */
     private static final long RECONNECT_LOG_THROTTLE_NANOS = 5_000_000_000L;
+    // True when this loop delta-encodes symbol dictionaries: it keeps a reconnect
+    // catch-up mirror (sentDict*) and re-registers the whole dictionary on a fresh
+    // server before replaying delta frames. See the sentDict* fields in the mutable
+    // section for the full mechanism. Gates all the delta-dict state.
+    private final boolean deltaDictEnabled;
     // Pre-converted to nanos for the comparison in sendDurableAckKeepaliveIfDue.
     // Zero or negative disables the keepalive entirely.
     private final long durableAckKeepaliveIntervalNanos;
@@ -211,51 +216,6 @@ public final class CursorWebSocketSendLoop implements QuietCloseable {
     private final long reconnectMaxDurationMillis;
     private final WebSocketResponse response = new WebSocketResponse();
     private final ResponseHandler responseHandler = new ResponseHandler();
-    // Delta symbol dictionary catch-up state (see swapClient). Active in memory
-    // mode, and in disk mode whenever the per-slot persisted dictionary opened --
-    // fresh slots included, not just recovered / orphan-drained ones. On a
-    // recovered / orphan-drained slot the constructor additionally SEEDS sentDict*
-    // from that persisted dictionary; a fresh slot starts with an empty mirror and
-    // grows it as frames are sent.
-    // deltaDictEnabled gates all of it. The loop mirrors, in sentDict*, every
-    // symbol it has ever sent -- the concatenated [len varint][utf8] bytes in
-    // global-id order (sentDictBytes*) plus the count (sentDictCount) -- so that
-    // on reconnect it can re-register the whole dictionary on the fresh server
-    // (which discards its dictionary on every disconnect) before replaying frames
-    // whose deltas start above id 0. All of this is touched only by the I/O thread.
-    // Footprint note: this mirror is a SECOND copy of the dictionary -- the same
-    // symbols the producer's GlobalSymbolDictionary already holds as Java Strings --
-    // kept as native UTF-8 bytes for the reconnect-catch-up capability. So a
-    // memory-mode connection's steady-state dictionary footprint is ~2x the symbol
-    // set. It is bounded by distinct-symbol count (not per-row) and never trimmed
-    // 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;
-    // 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
-    // consumes wire sequences (nextWireSeq), so nextWireSeq > 0 no longer implies
-    // "the head frame was sent": onClose's poison-strike gate and
-    // handleServerRejection's pre-send gate key off THIS instead. Without it, a
-    // transient outage AFTER the catch-up but BEFORE the first data frame (a
-    // flapping LB/middlebox that accepts 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.
-    private boolean dataFrameSentThisConnection;
-    private long sentDictBytesAddr;
-    private int sentDictBytesCapacity;
-    private int sentDictBytesLen;
-    private int sentDictCount;
-    // End position (native address) written by the last readVarintAt() call.
-    private long varintEnd;
     private final CountDownLatch shutdownLatch = new CountDownLatch(1);
     private final AtomicLong totalAcks = new AtomicLong();
     // Counters for observability of the durable-ack path. Both are zero
@@ -279,6 +239,49 @@ public final class CursorWebSocketSendLoop implements QuietCloseable {
     // by category. Includes both retriable and terminal outcomes — i.e. every
     // server-side rejection observed regardless of how the loop reacted.
     private final AtomicLong totalServerErrors = new AtomicLong();
+    // Delta symbol dictionary catch-up state (see swapClient, deltaDictEnabled).
+    // Active in memory mode, and in disk mode whenever the per-slot persisted
+    // dictionary opened -- fresh slots included, not just recovered / orphan-drained
+    // ones. On a recovered / orphan-drained slot the constructor additionally SEEDS
+    // sentDict* from that persisted dictionary; a fresh slot starts with an empty
+    // mirror and grows it as frames are sent. The loop mirrors, in sentDict*, every
+    // symbol it has ever sent -- the concatenated [len varint][utf8] bytes in
+    // global-id order (sentDictBytes*) plus the count (sentDictCount) -- so that on
+    // reconnect it can re-register the whole dictionary on the fresh server (which
+    // discards its dictionary on every disconnect) before replaying frames whose
+    // deltas start above id 0. All of this is touched only by the I/O thread.
+    // Footprint note: this mirror is a SECOND copy of the dictionary -- the same
+    // symbols the producer's GlobalSymbolDictionary already holds as Java Strings --
+    // kept as native UTF-8 bytes for the reconnect-catch-up capability. So a
+    // memory-mode connection's steady-state dictionary footprint is ~2x the symbol
+    // set. It is bounded by distinct-symbol count (not per-row) and never trimmed 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 long sentDictBytesAddr;
+    private int sentDictBytesCapacity;
+    private int sentDictBytesLen;
+    private int sentDictCount;
+    // 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 consumes
+    // wire sequences (nextWireSeq), so nextWireSeq > 0 no longer implies "the head
+    // frame was sent": onClose's poison-strike gate and handleServerRejection's
+    // pre-send gate key off THIS instead. Without it, a transient outage AFTER the
+    // catch-up but BEFORE the first data frame (a flapping LB/middlebox that accepts
+    // 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.
+    private boolean dataFrameSentThisConnection;
+    // End position (native address) written by the last readVarintAt() call.
+    private long varintEnd;
     private WebSocketClient client;
     // Optional: when non-null, every server-rejection error (retriable and
     // terminal alike) is offered to the dispatcher for async delivery to the user's

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

@@ -606,6 +606,62 @@ public long maxSymbolDeltaEnd(int headerMagic, int flagsOffset, int flagDeltaMas
         return maxEnd;
     }
 
+    /**
+     * Highest {@code deltaStart} any delta-flagged frame at or below
+     * {@code maxFsnInclusive} carries, or {@code 0} when NO delta-flagged frame is
+     * present. A result of {@code 0} means every surviving symbol-bearing frame is
+     * SELF-SUFFICIENT -- it re-registers its dictionary from id 0 (a full-dict
+     * frame), so the slot replays with no persisted dictionary at all; a result
+     * {@code > 0} means at least one frame is a non-self-sufficient delta whose ids
+     * depend on a prior frame's (or the persisted dictionary's) registrations.
+     * {@link CursorSendEngine} pairs this with {@link #maxSymbolDeltaEnd} at
+     * recovery: when the persisted {@code .symbol-dict} is a subset of the ids the
+     * frames reference BUT this returns {@code 0}, the slot was written in full-dict
+     * fallback (the dictionary never opened when writing) and recovers in full-dict
+     * mode instead of failing clean; a torn DELTA dictionary has a {@code
+     * deltaStart > 0} frame and still fails clean. Same read-only frame walk and
+     * layout as {@link #maxSymbolDeltaEnd}; only frames at or below {@code
+     * maxFsnInclusive} count (the retired orphan-deferred tail is excluded).
+     */
+    public long maxSymbolDeltaStart(int headerMagic, int flagsOffset, int flagDeltaMask, int qwpHeaderSize, long maxFsnInclusive) {
+        long maxStart = 0L;
+        long off = HEADER_SIZE;
+        long frames = frameCount;
+        for (long i = 0; i < frames; i++) {
+            long fsn = baseSeq + i;
+            int payloadLen = Unsafe.getUnsafe().getInt(mmapAddress + off + 4);
+            long payload = mmapAddress + off + FRAME_HEADER_SIZE;
+            // Skip the orphan-deferred tail (fsn > maxFsnInclusive): retired, never
+            // sent, so its baseline must not count. off still advances below.
+            if (fsn <= maxFsnInclusive
+                    && payloadLen >= qwpHeaderSize
+                    && payloadLen > flagsOffset
+                    && Unsafe.getUnsafe().getInt(payload) == headerMagic
+                    && (Unsafe.getUnsafe().getByte(payload + flagsOffset) & flagDeltaMask) != 0) {
+                long p = payload + qwpHeaderSize;
+                long limit = payload + payloadLen;
+                long deltaStart = 0L;
+                int shift = 0;
+                while (p < limit) {
+                    byte b = Unsafe.getUnsafe().getByte(p++);
+                    deltaStart |= (long) (b & 0x7F) << shift;
+                    if ((b & 0x80) == 0) {
+                        break;
+                    }
+                    shift += 7;
+                    if (shift > 35) {
+                        break; // corrupt run; recovered frames are CRC-valid, so defensive only
+                    }
+                }
+                if (deltaStart > maxStart) {
+                    maxStart = deltaStart;
+                }
+            }
+            off += FRAME_HEADER_SIZE + payloadLen;
+        }
+        return maxStart;
+    }
+
     /**
      * Number of frames written since {@link #create} (or recovered by
      * {@link #openExisting}). Used by {@code SegmentRing} to compute

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

@@ -566,6 +566,25 @@ public synchronized long maxSymbolDeltaEnd(int headerMagic, int flagsOffset, int
         return Math.max(maxEnd, end);
     }
 
+    /**
+     * Highest {@code deltaStart} any symbol-dict delta frame at or below
+     * {@code maxFsnInclusive} carries (0 when none, or when every such frame is
+     * self-sufficient). See {@link MmapSegment#maxSymbolDeltaStart}; paired with
+     * {@link #maxSymbolDeltaEnd} at recovery to tell a full-dict-fallback slot
+     * (recoverable with no dictionary) from a torn delta dictionary (fails clean).
+     */
+    public synchronized long maxSymbolDeltaStart(int headerMagic, int flagsOffset, int flagDeltaMask, int qwpHeaderSize, long maxFsnInclusive) {
+        long maxStart = 0L;
+        for (int i = 0, n = sealedSegments.size(); i < n; i++) {
+            long start = sealedSegments.get(i).maxSymbolDeltaStart(headerMagic, flagsOffset, flagDeltaMask, qwpHeaderSize, maxFsnInclusive);
+            if (start > maxStart) {
+                maxStart = start;
+            }
+        }
+        long start = active.maxSymbolDeltaStart(headerMagic, flagsOffset, flagDeltaMask, qwpHeaderSize, maxFsnInclusive);
+        return Math.max(maxStart, start);
+    }
+
     public MmapSegment getActive() {
         return active;
     }