Rebuild the dictionary from the frames still on disk

A slot whose symbol dictionary was damaged was being condemned as
unreplayable while every symbol it needed was sitting on disk. The
sender told the user to resend data that had never been lost, and after
the previous commit it set the slot aside instead.

Three things kept the rebuild from ever running.

setCursorEngine gated the producer's seed on deltaDictEnabled. That flag
is false exactly when the slot's dictionary failed to open -- fd
exhaustion, a read-only remount, ENOSPC -- which is precisely when the
frames on disk are the only surviving copy of the symbols. The rebuild
was dead code for the very case it was written for.

collectReplaySymbolsAbove walked from ackedFsn+1, so it skipped the
acked frames. But an acked frame is not trimmed the instant it is
acked: trim unlinks whole SEALED segments. The frames that registered
the early ids are almost always still there, and they are exactly the
ids the replay set's deltas start above. It walks every frame on disk
now, and a slot whose registering frames really HAVE been trimmed still
reports the gap, because then nothing holds them.

And the send loop's mirror seeded only from the dictionary file, never
from the frames -- so even with the other two fixed, the loop would
still condemn (deltaStart > sentDictCount) a slot the producer had just
seeded successfully. The mirror is the loop's model of what the SERVER
holds and it is what the catch-up ships, so it now seeds from the same
two sources in the same order, and the two land on the same number by
construction. Only when a surviving frame actually depends on ids it
does not carry: at maxSymbolDeltaStart == 0 every frame re-registers
from id 0 as it replays, so a full-dict slot stays catch-up-free.

The recovery tests now assert the strong form -- the fresh server's
reconstructed dictionary must come back COMPLETE and in order, which
can only happen if the catch-up carried the ids the acked frames still
hold. A null-padded or shifted dictionary fails there, which is the
corruption the old guard condemned these slots to avoid. It never had
to.

The quarantine now covers only what is genuinely unrecoverable: the
registering frames trimmed away and the dictionary torn, so nothing
anywhere holds those ids.

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

@@ -2258,7 +2258,11 @@ public void setCursorEngine(CursorSendEngine engine, boolean takeOwnership) {
         // persisted dictionary so newly ingested symbols continue from the
         // recovered ids (rather than colliding with them at 0), and the delta
         // baseline resumes where the crashed session left off.
-        if (deltaDictEnabled && engine.wasRecoveredFromDisk()) {
+        // NOT gated on deltaDictEnabled. That flag is false exactly when the slot's dictionary
+        // failed to open -- which is precisely when the frames on disk are the only surviving
+        // copy of the symbols and the rebuild matters most. Gating the seed on it made the
+        // rebuild dead code for the very case it was written for.
+        if (engine != null && engine.wasRecoveredFromDisk()) {
             seedGlobalDictionaryFromPersisted(engine.getPersistedSymbolDict());
         }
     }

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

@@ -108,6 +108,14 @@ public final class CursorSendEngine implements QuietCloseable {
     // against the recovered dictionary size to fail clean instead. Computed once
     // in the constructor's recovery branch; -1 elsewhere.
     private long recoveredMaxSymbolId = -1L;
+    // Highest deltaStart across the recovered COMMITTED frames; 0 when none carries a symbol
+    // dictionary. ZERO means every surviving frame is SELF-SUFFICIENT -- it re-registers its
+    // dictionary from id 0 -- so the slot replays with no dictionary at all and the send loop
+    // needs no catch-up. ABOVE zero means at least one frame is a true delta whose ids depend
+    // on registrations it does not itself carry, so the loop must seed its mirror (and ship a
+    // catch-up) before replaying. Both the full-dict-fallback discard below and the send
+    // loop's mirror seeding key off this.
+    private long recoveredMaxSymbolDeltaStart;
     // FSN of the last frame of a recovered orphaned deferred tail, or -1 when
     // the recovered ring has no such tail. When >= 0, frames
     // [recoveredCommitBoundaryFsn + 1 .. recoveredOrphanTipFsn] all carry
@@ -380,14 +388,15 @@ private CursorSendEngine(String sfDir, long segmentSizeBytes, SegmentManager man
                 // 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(
+                this.recoveredMaxSymbolDeltaStart = recovered.maxSymbolDeltaStart(
                         QwpConstants.MAGIC_MESSAGE,
                         QwpConstants.HEADER_OFFSET_FLAGS,
                         QwpConstants.FLAG_DELTA_SYMBOL_DICT,
                         QwpConstants.HEADER_SIZE,
-                        recoveredCommitBoundaryFsn) == 0L) {
+                        recoveredCommitBoundaryFsn);
+                if (persistedDictInProgress != null
+                        && recoveredMaxSymbolId >= persistedDictInProgress.size()
+                        && recoveredMaxSymbolDeltaStart == 0L) {
                     persistedDictInProgress.close();
                     persistedDictInProgress = null;
                 }
@@ -715,7 +724,14 @@ public long collectReplaySymbolsAbove(int baseline, ObjList<String> out) {
                 QwpConstants.HEADER_OFFSET_FLAGS,
                 QwpConstants.FLAG_DELTA_SYMBOL_DICT,
                 QwpConstants.HEADER_SIZE,
-                ackedFsn() + 1L,
+                // Walk from the LOWEST frame still on disk, not from ackedFsn+1. An acked frame
+                // is not trimmed the instant it is acked -- trim drops whole SEALED segments --
+                // so the symbols it registered are usually still sitting right there, and they
+                // are exactly the ids the replay set's deltas start above. Skipping them threw
+                // away the only surviving copy of those symbols and condemned a slot that had
+                // everything it needed. A slot whose registering frames really HAVE been
+                // trimmed still reports the gap, because then no frame on disk holds them.
+                0L,
                 recoveredCommitBoundaryFsn,
                 baseline,
                 out
@@ -842,6 +858,20 @@ public long recoveredMaxSymbolId() {
         return recoveredMaxSymbolId;
     }
 
+    /**
+     * Highest {@code deltaStart} across the recovered committed frames; {@code 0} when every
+     * surviving frame is self-sufficient (or none carries a dictionary at all).
+     * <p>
+     * The send loop uses this to decide whether it needs a catch-up: at zero, every frame
+     * re-registers its dictionary from id 0 as it replays, so seeding the mirror -- and
+     * shipping a catch-up frame off it -- would be pure redundancy. Above zero, at least one
+     * frame's delta starts above ids it does not itself carry, so the mirror must hold those
+     * ids before the replay begins or the server null-pads the hole.
+     */
+    public long recoveredMaxSymbolDeltaStart() {
+        return recoveredMaxSymbolDeltaStart;
+    }
+
     /**
      * FSN of the last frame of a recovered orphaned deferred tail, or
      * {@code -1} when none. See {@link #recoveredCommitBoundaryFsn()}: the

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

@@ -43,6 +43,8 @@
 import io.questdb.client.std.CharSequenceLongHashMap;
 import io.questdb.client.std.MemoryTag;
 import io.questdb.client.std.QuietCloseable;
+import io.questdb.client.std.str.Utf8s;
+import io.questdb.client.std.ObjList;
 import io.questdb.client.std.Unsafe;
 import org.jetbrains.annotations.TestOnly;
 import org.slf4j.Logger;
@@ -699,6 +701,25 @@ public CursorWebSocketSendLoop(WebSocketClient client, CursorSendEngine engine,
                 sentDictCount = pd.size();
             }
         }
+        // ...and then from the surviving frames' own delta sections, exactly as the producer
+        // seeds its dictionary. The mirror is the loop's model of what the SERVER holds and it
+        // is what the catch-up frame ships, so when it stops at the persisted prefix while the
+        // producer's baseline runs past it, the loop condemns (deltaStart > sentDictCount) a
+        // slot the producer just seeded successfully. Same two sources, same order, so the two
+        // land on the same number by construction.
+        // ...but ONLY when a surviving frame actually depends on ids it does not carry
+        // (recoveredMaxSymbolDeltaStart > 0). At zero every frame is self-sufficient and
+        // 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.
+        if (engine.recoveredMaxSymbolDeltaStart() > 0L) {
+            ObjList<String> fromFrames = new ObjList<>();
+            if (engine.collectReplaySymbolsAbove(sentDictCount, fromFrames) >= 0) {
+                for (int i = 0, n = fromFrames.size(); i < n; i++) {
+                    appendSymbolToMirror(fromFrames.getQuick(i));
+                }
+            }
+        }
         this.fsnAtZero = fsnAtZero;
         this.parkNanos = parkNanos;
         this.reconnectFactory = reconnectFactory;
@@ -2268,6 +2289,23 @@ private void accumulateSentDict(long payloadAddr, int payloadLen, int deltaStart
         sentDictCount += (int) newCount;
     }
 
+    /**
+     * Appends one symbol to the sent-dictionary mirror in wire form
+     * ({@code [len varint][utf8]}), advancing {@code sentDictCount}. Seeds the mirror at
+     * construction; {@link #accumulateSentDict} extends it from live frames thereafter.
+     */
+    private void appendSymbolToMirror(CharSequence symbol) {
+        int utf8Len = Utf8s.utf8Bytes(symbol);
+        int wireLen = NativeBufferWriter.varintSize(utf8Len) + utf8Len;
+        ensureSentDictCapacity((long) sentDictBytesLen + wireLen);
+        long p = NativeBufferWriter.writeVarint(sentDictBytesAddr + sentDictBytesLen, utf8Len);
+        if (utf8Len > 0) {
+            Utf8s.strCpyUtf8(symbol, p, utf8Len);
+        }
+        sentDictBytesLen += wireLen;
+        sentDictCount++;
+    }
+
     private void ensureSentDictCapacity(long required) {
         if (sentDictBytesCapacity >= required) {
             return;