Recover slots whose symbol dictionary is damaged

The QWP store-and-forward recovery path treated a damaged .symbol-dict as
unrecoverable in two places, stranding slots that are in fact replayable.
Both stem from keying on the dictionary FILE rather than on what the
surviving frames themselves define.

CursorWebSocketSendLoop ran the torn-dictionary guard unconditionally but
gated accumulateSentDict, the reconnect catch-up and the mirror seed on
engine.isDeltaDictEnabled(). That flag goes false exactly when a disk
slot's dictionary fails to open -- fd exhaustion, a read-only remount,
ENOSPC -- which is precisely when the frames already on disk are still
deltas. With the mirror gated off, sentDictCount froze at 0 while the
ungated guard kept comparing against it, so the frame at deltaStart=1
latched a terminal and the background drainer quarantined the slot behind
a .failed sentinel. Yet replaying that sequence from id 0 rebuilds the
dictionary on the server contiguously and drains perfectly. The loop now
always mirrors, always catches up and always guards: the mirror is its
model of what the SERVER holds, so it must track every mode.
isDeltaDictEnabled() now governs only what the producer emits and
persists.

seedGlobalDictionaryFromPersisted threw whenever the surviving frames
out-reached the persisted dictionary. That bricked Sender.build() for
slots the drainer replays fine, and permanently: build()'s catch releases
the slot lock, but the sender that would have hosted the orphan drainer is
the one that just failed, so the retry recovers the same slot and throws
again. Relaxing the guard would have been worse than the bug -- with an
empty dictionary the producer hands its next symbol id 0, an id the frames
already define, and accumulateSentDict skips it, so the next catch-up
replays the stale symbol and misattributes it silently. The producer now
seeds from the dictionary's intact prefix AND THEN from the surviving
frames' own delta sections, through collectReplaySymbolsAbove
(MmapSegment -> SegmentRing -> CursorSendEngine). Those are the same two
sources, in the same order, that the send loop's mirror is built from, so
the producer's delta baseline and the mirror's coverage land on the same
number by construction.

A genuine gap -- ids introduced by frames since acked and trimmed away --
still fails clean, and now on exactly the condition the send loop's replay
guard trips on, so producer and drainer agree on which slots are
recoverable.

The two cap-gap settle-budget tests derived their retriable loop bound
from MAX_CATCHUP_CAP_GAP_ATTEMPTS itself, so a regression of that constant
to 1 -- the very bug they name -- ran the loop zero times and let the test
pass green. They now pin the budget against a literal.

Tests: an untrimmed unopenable-dictionary slot must replay gap-free; a
torn (subset) and a wholly lost dictionary must rebuild from the surviving
frames and place a new symbol above the recovered tip. The two
watermark-stamped siblings still fail clean.

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

@@ -3847,91 +3847,88 @@ private void resetSymbolDictStateForNewConnection() {
     }
 
     /**
-     * On recovery, repopulates the producer's {@link GlobalSymbolDictionary} from
-     * the slot's persisted dictionary (ids assigned in the same ascending order,
-     * so they match the recovered frames) and resumes the delta baseline at the
-     * recovered tip, so newly ingested symbols continue above the recovered ids.
+     * On recovery, repopulates the producer's {@link GlobalSymbolDictionary} so that newly
+     * ingested symbols continue ABOVE every id the surviving frames already define, and
+     * resumes the delta baseline at that tip.
      * <p>
-     * Uses {@link GlobalSymbolDictionary#addRecoveredSymbol} (append, NOT de-dup):
-     * the persisted dictionary, the on-wire delta and the send-loop catch-up mirror
-     * all key on the entry POSITION (id), so the producer id space must match the
-     * persisted entry count exactly. {@code getOrAddSymbol} would collapse two
-     * source strings that decode to the same characters -- only malformed lone
-     * UTF-16 surrogates, which UTF-8-encode to {@code '?'} -- leaving this
-     * dictionary shorter than {@code pd.size()} and desyncing
-     * {@code sentMaxSymbolId} from the mirror's {@code sentDictCount = pd.size()},
-     * which silently misattributes later symbols after a reconnect.
+     * Seeds from TWO sources, in this order:
+     * <ol>
+     *   <li>the slot's persisted {@code .symbol-dict} -- its intact prefix; then</li>
+     *   <li>the surviving frames' OWN delta sections, for every id above that prefix
+     *       ({@link CursorSendEngine#collectReplaySymbolsAbove}).</li>
+     * </ol>
+     * Those are exactly the two sources, in exactly the order, that the send loop's mirror
+     * is built from: its constructor seeds {@code sentDictCount} from the same dictionary,
+     * and {@code accumulateSentDict} then extends it from the same frames as they replay. So
+     * the producer's {@code sentMaxSymbolId + 1} and the loop's {@code sentDictCount} land on
+     * the same number BY CONSTRUCTION -- the invariant the torn-dictionary guard rests on --
+     * rather than by the two happening to agree.
      * <p>
-     * <b>Host-crash tear guard.</b> The persisted dictionary is NOT fsync'd (see
-     * {@code PersistedSymbolDict}), so a host/power crash can lose its
-     * most-recently-written (highest-id) entries while the segment frames that
-     * introduced those ids survive -- and those newest frames, being the least
-     * likely to be acked, replay on recovery. The send loop's catch-up mirror then
-     * rebuilds the missing ids from those frames' own delta bytes, but THIS producer
-     * -- seeded only from the shorter dictionary -- would assign its next new symbol
-     * an id the surviving frames already define, putting two symbols on one id and
-     * silently misattributing values. The send loop's replay guard only catches a
-     * GAP ({@code deltaStart > sentDictCount}); a frame that introduces exactly the
-     * torn-off id ({@code deltaStart == pd.size()}) slips through and self-heals the
-     * 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.
+     * Uses {@link GlobalSymbolDictionary#addRecoveredSymbol} (append, NOT de-dup): the
+     * persisted dictionary, the on-wire delta and the mirror all key on the entry POSITION
+     * (id), so the producer's id space must match the recovered entry count exactly.
+     * {@code getOrAddSymbol} would collapse two source strings that decode to the same
+     * characters -- only malformed lone UTF-16 surrogates, which UTF-8-encode to {@code '?'}
+     * -- leaving this dictionary SHORTER than the count and silently misattributing later
+     * symbols.
      * <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.
+     * <b>Why seeding from the frames matters.</b> The dictionary is not fsync'd (see
+     * {@code PersistedSymbolDict}), so a host/power crash can tear off its newest entries
+     * while the segment frames that introduced those ids survive -- and those newest frames,
+     * being the least likely to be acked, are exactly the ones that replay. Seeded from the
+     * short dictionary alone, this producer would hand its next new symbol an id those frames
+     * already define, putting two symbols on one id and silently misattributing values. The
+     * old code detected that and threw, which was safe but far too blunt: it bricked
+     * {@code build()} for slots the background drainer replays PERFECTLY, because the frames
+     * carry the torn-off symbols in their own deltas and {@code accumulateSentDict} rebuilds
+     * the dictionary from them. This method now rebuilds the producer from the same bytes,
+     * so a torn -- or entirely lost -- dictionary is recoverable whenever the surviving
+     * frames define the ids themselves. The next flush's write-ahead persist then re-writes
+     * those ids (it resumes from {@code pd.size()}), healing the side-file on disk.
+     * <p>
+     * <b>What still fails clean.</b> A genuine GAP: the ids below a surviving frame's delta
+     * start were introduced by frames that were acked and TRIMMED away, so they lived only in
+     * the lost dictionary and nothing can rebuild them.
+     * {@code collectReplaySymbolsAbove} returns -1 for that and we throw. It is the same
+     * condition the send loop's replay guard ({@code deltaStart > sentDictCount}) trips on, so
+     * producer and drainer now agree on exactly which slots are recoverable, instead of the
+     * producer rejecting slots the drainer drains.
      */
     private void seedGlobalDictionaryFromPersisted(PersistedSymbolDict pd) {
-        if (pd == null) {
+        if (cursorEngine == null) {
             return;
         }
-        // Run the torn-dictionary guard BEFORE the empty-dictionary short-circuit
-        // 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()) {
+        // 1. The dictionary's intact prefix. addRecoveredSymbol appends without de-dup, so
+        //    the producer's size tracks pd.size() exactly -- which is what the send loop's
+        //    mirror also seeds sentDictCount from.
+        int baseline = 0;
+        if (pd != null && pd.size() > 0) {
+            ObjList<String> persisted = pd.readLoadedSymbols();
+            for (int i = 0, n = persisted.size(); i < n; i++) {
+                globalSymbolDictionary.addRecoveredSymbol(persisted.getQuick(i));
+            }
+            baseline = globalSymbolDictionary.size();
+        }
+        // 2. Everything the surviving frames define above that prefix, straight out of their
+        //    own delta sections -- the same bytes, in the same order, accumulateSentDict will
+        //    feed the mirror as those frames go back on the wire.
+        ObjList<String> fromFrames = new ObjList<>();
+        long coverage = cursorEngine.collectReplaySymbolsAbove(baseline, fromFrames);
+        if (coverage < 0) {
             throw new LineSenderException(
-                    "recovered store-and-forward symbol dictionary is a subset of the surviving frames "
-                            + "(likely a host crash tore its unsynced tail): frames reference symbol id "
-                            + cursorEngine.recoveredMaxSymbolId() + " but the recovered dictionary holds only "
-                            + pd.size() + " id(s); resuming would reuse ids the frames already define -- "
-                            + "resend the affected data");
-        }
-        if (pd.size() == 0) {
-            return;
-        }
-        ObjList<String> symbols = pd.readLoadedSymbols();
-        for (int i = 0, n = symbols.size(); i < n; i++) {
-            globalSymbolDictionary.addRecoveredSymbol(symbols.getQuick(i));
-        }
+                    "recovered store-and-forward symbol dictionary is incomplete and cannot be "
+                            + "rebuilt from the surviving frames (likely a host crash tore its unsynced "
+                            + "tail): the frames reference symbol ids below their own delta start, which "
+                            + "were introduced by frames since acked and trimmed away, so nothing still "
+                            + "holds them; the recovered dictionary holds only "
+                            + (pd == null ? 0 : pd.size()) + " id(s) -- resend the affected data");
+        }
+        for (int i = 0, n = fromFrames.size(); i < n; i++) {
+            globalSymbolDictionary.addRecoveredSymbol(fromFrames.getQuick(i));
+        }
+        // Producer baseline == the coverage the replay will establish == the mirror's
+        // sentDictCount once those frames have gone out. The first new frame therefore
+        // starts its delta exactly at the tip, and the replay guard passes.
         sentMaxSymbolId = globalSymbolDictionary.size() - 1;
     }
 

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

@@ -661,6 +661,44 @@ public synchronized void close() {
         }
     }
 
+    /**
+     * Rebuilds, from the surviving frames' OWN delta sections, the symbols the upcoming
+     * replay will register ABOVE {@code baseline}, appending them to {@code out} in
+     * ascending id order. Returns the coverage the replay establishes (one past the highest
+     * id it registers), or {@code -1} when the frames have a genuine GAP above the baseline.
+     * <p>
+     * The producer seeds its dictionary from the persisted {@code .symbol-dict} and THEN
+     * from this, so it is fed the same entries, in the same order, that the send loop's
+     * {@code accumulateSentDict} will feed its mirror from as those very frames replay. The
+     * producer's delta baseline and the loop's mirror coverage therefore land on the same
+     * number by construction, which is the invariant the torn-dictionary guard rests on.
+     * <p>
+     * This is what makes a slot whose dictionary was torn -- or lost outright -- still
+     * recoverable when the surviving frames define the ids themselves (they carry the
+     * symbols in their own deltas; it is why the orphan drainer can drain such a slot). Only
+     * a real gap, where the ids were introduced by frames since acked and trimmed away, is
+     * unrecoverable -- and that is precisely when this returns -1.
+     * <p>
+     * Bounded above by {@link #recoveredCommitBoundaryFsn} like {@link #recoveredMaxSymbolId}:
+     * frames past it are the aborted orphan-deferred tail, retired without ever being
+     * transmitted, so their ids never reach a server and must not inflate the baseline.
+     */
+    public long collectReplaySymbolsAbove(int baseline, ObjList<String> out) {
+        if (ring == null) {
+            return baseline;
+        }
+        return ring.collectReplaySymbolsAbove(
+                QwpConstants.MAGIC_MESSAGE,
+                QwpConstants.HEADER_OFFSET_FLAGS,
+                QwpConstants.FLAG_DELTA_SYMBOL_DICT,
+                QwpConstants.HEADER_SIZE,
+                ackedFsn() + 1L,
+                recoveredCommitBoundaryFsn,
+                baseline,
+                out
+        );
+    }
+
     /**
      * Pass-through to {@link SegmentRing#findSegmentContaining(long)}.
      */