Close four data-integrity holes in the delta symbol dictionary

C1. CursorSendEngine.close() unlinked the parent-anchored logical slot
lock while Sender.build() still held it. A fresh slot is "fully drained"
by definition (publishedFsn() < 0), and build() closes the engine from
inside its acquireLogical scope whenever connect fails -- the ordinary
"server isn't up yet" startup. On POSIX the unlink frees the pathname
without releasing the flock, so the next acquireLogical creates a second
inode and locks it successfully: two parties owning the lock that exists
solely to serialise the quarantine close->rename->recreate window.
close(boolean) lets a caller that holds the lock skip the reclaim;
build() and quarantineTornSlot pass false. An unheld lock is still
reclaimed, so .slot-locks does not grow without bound.

C2. Recovery rebuilds the producer dictionary from the side-file's
intact prefix AND the surviving frames' own deltas, then resumes
sentMaxSymbolId at the combined tip -- above pd.size(). Every frame
published afterwards then carried a deltaStart the side-file could not
describe. The steady-state write-ahead did not close that gap: it
persists [pd.size() .. currentBatchMaxSymbolId] and returns early when
the batch's highest id is below pd.size(), so it healed only if, and
only as far as, later traffic referenced the recovered high ids. The
frames carrying those ids are the oldest unacked and so the first to be
acked and trimmed; once gone, an ordinary process crash -- which
store-and-forward promises to survive -- left a slot whose frames
reference ids nothing holds, and build() quarantined it. Recovery now
heals the side-file eagerly and in full, before any new frame publishes.

That heal advances pd.size() before the send loop is built, and the loop
seeded sentDictCount from the live size() while taking its bytes from
the fixed loaded region -- so the mirror would have claimed symbols it
did not hold. PersistedSymbolDict now exposes recoveredSize(), the entry
count that matches loadedEntriesAddr/Len, and the loop and the loaded-
region decoder both key off that instead.

C3. deltaDictEnabled was written once at setCursorEngine with no runtime
disable, so a mid-run .symbol-dict write failure killed flush() forever.
A full disk reaches exactly that state: SF's segments are pre-allocated
mmap files and stay writable while the growing side-file does not. Full
self-sufficient frames need no side file at all, so a persist failure
now degrades to them. The failing flush still throws -- beginMessage has
already baked a delta deltaStart into the staged frame -- but the throw
precedes every publish, so the rows stay buffered and the next flush
re-encodes them from id 0.

C4. Two recovery-seed exits threw raw IllegalStateException, which
Sender.build() does not route to its quarantine handler: with a stable
senderId every restart re-recovered the same slot and rethrew, so the
application could never construct a Sender and could not even buffer.
Both now throw UnreplayableSlotException. QwpWebSocketSender.connect's
rollback also let a close() failure replace the original exception and
demote a recoverable slot back to that brick; it now addSuppressed.

Regression tests, each verified to fail with its fix reverted:
EngineCloseSlotLockReleaseTest#testCloseDoesNotUnlinkALogicalLockItsCallerHolds
("close(false) must leave the logical lock file alone") and
DeltaDictRecoveryTest#testRecoveryHealsThePersistedDictionaryBeforeAnyNewFrame
("expected:<3> but was:<2>"). A companion test pins that an unheld
logical lock is still reclaimed. Full client suite: 2687 tests green.

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

Author: glasstiger

core/src/main/java/io/questdb/client/Sender.java

@@ -1643,7 +1643,8 @@ public Sender build() {
                             // rather than loop.
                             if (quarantined || slotPath == null) {
                                 try {
-                                    cursorEngine.close();
+                                    // close(false): we still hold the logical slot lock.
+                                    cursorEngine.close(false);
                                 } catch (Throwable ignored) {
                                     // best-effort
                                 }
@@ -1655,9 +1656,12 @@ public Sender build() {
                                     actualSfMaxTotalBytes, actualSfAppendDeadlineNanos);
                         } catch (Throwable t) {
                             // connect() failed before ownership of cursorEngine
-                            // transferred — close it ourselves.
+                            // transferred — close it ourselves. close(false)
+                            // because logicalSlotLock is still held here: a fresh
+                            // slot is fully drained, so the default close would
+                            // unlink the very lock file this scope holds.
                             try {
-                                cursorEngine.close();
+                                cursorEngine.close(false);
                             } catch (Throwable ignored) {
                                 // best-effort
                             }
@@ -3064,8 +3068,10 @@ private static CursorSendEngine quarantineTornSlot(
             final String detail = cause.getMessage();
             // Release the slot lock and the dictionary fd before renaming. connect()'s failure
             // path already closed the engine; close() is idempotent, so make it explicit rather
-            // than depend on that.
-            torn.close();
+            // than depend on that. close(false): build() holds the logical slot lock across this
+            // whole transition -- that is precisely what serialises the rename against a queued
+            // orphan drainer -- so the engine must not unlink it.
+            torn.close(false);
             Runnable hook = quarantineAfterCloseHook;
             if (hook != null) {
                 hook.run();

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

@@ -804,7 +804,16 @@ public static QwpWebSocketSender connect(
             }
             sender.ensureConnected();
         } catch (Throwable t) {
-            sender.close();
+            // Preserve t's IDENTITY through the rollback. Sender.build() routes on the
+            // exception type -- only UnreplayableSlotException reaches its quarantine
+            // handler -- and close() accumulates cleanup errors and ends in
+            // rethrowTerminal, so letting a close failure propagate here would REPLACE t
+            // and silently demote a recoverable slot back to the permanent build() brick.
+            try {
+                sender.close();
+            } catch (Throwable closeFailure) {
+                t.addSuppressed(closeFailure);
+            }
             throw t;
         }
         return sender;
@@ -3864,6 +3873,84 @@ private void advanceSentMaxSymbolId() {
         }
     }
 
+    /**
+     * Stops emitting delta dictionaries for the rest of this sender's life, after the
+     * per-slot {@code .symbol-dict} has proved unwritable.
+     * <p>
+     * The side-file can stop accepting appends mid-run -- a full disk or an exhausted
+     * quota, where SF's own segments stay writable because they are pre-allocated mmap
+     * files while the dictionary is the one thing still growing. Without a way back,
+     * {@code deltaDictEnabled} is written once at {@code setCursorEngine} and every
+     * later {@code flush()} re-throws forever: a condition store-and-forward is built
+     * to survive becomes total, permanent ingestion loss.
+     * <p>
+     * Full self-sufficient frames need no side file at all -- each carries the whole
+     * dictionary from id 0, which is exactly what recovery and orphan-drain replay
+     * against a fresh server. So degrade instead of dying. The producer's monotonic
+     * baseline stops being consulted ({@link #symbolDeltaBaseline()} returns -1), and
+     * the write-ahead persist becomes a no-op.
+     * <p>
+     * Producer-thread only, like every other reader of {@code deltaDictEnabled}.
+     */
+    private void disableDeltaDict(Throwable cause) {
+        if (!deltaDictEnabled) {
+            return;
+        }
+        deltaDictEnabled = false;
+        LOG.warn("symbol dictionary persistence failed; this sender has switched to full "
+                + "self-sufficient frames for the rest of its life (bandwidth cost only -- "
+                + "no data is at risk, and recovery replays such frames without a side file)",
+                cause);
+    }
+
+    /**
+     * Writes the ids the surviving frames contributed above the persisted prefix back
+     * into {@code .symbol-dict}, immediately, before any new frame can be published.
+     * <p>
+     * {@link #seedGlobalDictionaryFromPersisted} can rebuild the producer dictionary from
+     * TWO sources -- the side-file's intact prefix and the surviving frames' own delta
+     * sections -- and then resumes {@code sentMaxSymbolId} at the combined tip. When the
+     * frames contributed anything, that tip is ABOVE {@code pd.size()}, so every frame
+     * published from here on carries a {@code deltaStart} the side-file cannot describe.
+     * That breaks the write-ahead invariant the whole design rests on: the persisted
+     * dictionary must be a superset of every recoverable frame's references.
+     * <p>
+     * The steady-state write-ahead does NOT close that gap on its own. It persists
+     * {@code [pd.size() .. currentBatchMaxSymbolId]} and returns early when the batch's
+     * highest id is below {@code pd.size()}, so it heals only if -- and only as far as --
+     * a later batch happens to reference the recovered high ids. Meanwhile the frames
+     * that carry those ids are the oldest unacked, so they are the FIRST to be acked and
+     * trimmed. Once they are gone, an ordinary process crash (which store-and-forward
+     * promises to survive; only the original tear needs a host crash) leaves a slot whose
+     * frames reference ids nothing holds: recovery marks a gap and {@code build()}
+     * quarantines it with "resend the affected data".
+     * <p>
+     * Healing here, eagerly and in full, restores the invariant before the window opens.
+     */
+    private void healPersistedDictionary(PersistedSymbolDict pd) {
+        if (pd == null || !deltaDictEnabled) {
+            return;
+        }
+        int from = pd.size();
+        int to = globalSymbolDictionary.size() - 1;
+        if (to < from) {
+            return; // the side-file already covers everything the frames defined
+        }
+        try {
+            pd.appendSymbols(globalSymbolDictionary, from, to);
+        } catch (Throwable t) {
+            if (t instanceof Error) {
+                throw (Error) t;
+            }
+            // Do NOT fail recovery: the surviving frames still carry these ids in their
+            // own deltas, so THIS session replays correctly either way. Only a future
+            // recovery, after those frames are trimmed, would be affected -- and the
+            // degrade below removes even that exposure by dropping back to frames that
+            // need no side file.
+            disableDeltaDict(t);
+        }
+    }
+
     /**
      * Appends the symbols this frame introduces ({@code [sentMaxSymbolId+1 ..
      * currentBatchMaxSymbolId]}) to the slot's persisted dictionary BEFORE the
@@ -3932,7 +4019,16 @@ private void persistNewSymbolsBeforePublish() {
             if (t instanceof Error) {
                 throw (Error) t;
             }
-            throw new LineSenderException("failed to persist symbol dictionary before publish", t);
+            // Degrade before throwing, so this failure is survivable rather than terminal:
+            // every LATER flush emits full self-sufficient frames, which need no side file
+            // (see disableDeltaDict). This one flush still has to fail -- beginMessage has
+            // already baked a delta deltaStart into the staged frame, and publishing it
+            // would put ids on the ring that the side-file cannot describe. The throw
+            // precedes every publish, so the caller's rows stay buffered and the next
+            // flush() re-encodes them from id 0.
+            disableDeltaDict(t);
+            throw new LineSenderException("failed to persist symbol dictionary before publish; "
+                    + "this sender has switched to full self-sufficient frames -- retry the flush", t);
         }
     }
 
@@ -4054,6 +4150,10 @@ private void seedGlobalDictionaryFromPersisted(PersistedSymbolDict pd) {
         // 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;
+        // ...but the baseline now sits ABOVE pd.size() whenever the frames contributed
+        // ids, so restore the write-ahead invariant right now rather than hoping a later
+        // batch reaches high enough to do it. See healPersistedDictionary.
+        healPersistedDictionary(pd);
     }
 
     /**

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

@@ -409,14 +409,13 @@ private CursorSendEngine(String sfDir, long segmentSizeBytes, SegmentManager man
                     // baseline 0 once pd is gone, and checkedRecoveryAnalysis rejects a
                     // baseline that disagrees with the fold. Discarding a dictionary that
                     // held entries (size > 0) therefore desynchronised the two and threw
-                    // IllegalStateException("recovery symbol baseline mismatch") out of
-                    // build(). That is NOT an UnreplayableSlotException, so build()'s
-                    // quarantine handler could not catch it and set the slot aside: with a
-                    // stable senderId every restart re-recovered the same slot and threw
-                    // again, so the application could never construct a Sender -- it could
-                    // not even BUFFER new rows. Exactly the outage quarantineTornSlot
-                    // exists to prevent, on a slot that is fully recoverable (its frames
-                    // carry their whole dictionary inline).
+                    // "recovery symbol baseline mismatch" out of build(). checkedRecoveryAnalysis
+                    // now raises that as an UnreplayableSlotException, so build() would at least
+                    // set the slot aside rather than rethrow forever -- but quarantining is
+                    // still the wrong outcome HERE, because this slot is fully recoverable
+                    // (its frames carry their whole dictionary inline). Re-folding avoids the
+                    // mismatch entirely instead of trading a permanent brick for a needless
+                    // quarantine plus a "resend the affected data" the operator does not owe.
                     //
                     // Reachable on one transient plus one crash: a session whose
                     // .symbol-dict fails to open (EIO, fd exhaustion, a Windows share
@@ -640,7 +639,24 @@ public long appendOrFsn(long payloadAddr, int payloadLen, long spinDeadlineNanos
     }
 
     @Override
-    public synchronized void close() {
+    public void close() {
+        close(true);
+    }
+
+    /**
+     * As {@link #close()}, but {@code reclaimLogicalSlotLock == false} skips the
+     * parent-anchored logical slot-lock unlink at fully-drained retirement.
+     * <p>
+     * A caller that HOLDS that lock must pass {@code false}. {@code Sender.build()}
+     * keeps it across engine construction and connect, and closes the engine from
+     * inside that scope when connect fails -- and a fresh slot is "fully drained" by
+     * definition ({@code publishedFsn() < 0}), so the default path would unlink the
+     * lock file while build() still holds the flock on it. On POSIX that frees the
+     * pathname without releasing the lock, so the next {@code acquireLogical} creates
+     * a SECOND inode and locks it successfully: two parties owning a lock whose only
+     * job is serialising the quarantine close-&gt;rename-&gt;recreate window.
+     */
+    public synchronized void close(boolean reclaimLogicalSlotLock) {
         if (closed) return;
         closed = true;
         // Capture drain state BEFORE closing the ring — once the ring is
@@ -728,16 +744,20 @@ public synchronized void close() {
                     PersistedSymbolDict.removeOrphan(sfDir);
                 } catch (Throwable ignored) {
                 }
-                try {
-                    // The logical slot lock lives OUTSIDE the slot dir (in the
-                    // shared .slot-locks dir) so it survives a slot rename; the
-                    // fully-drained retirement that removes this slot's other
-                    // side-files must remove it too, or .slot-locks accumulates a
-                    // dead lock+pid pair per distinct slot name for the lifetime of
-                    // sf_dir. This engine still holds the directory-local lock, so
-                    // the best-effort unlink is safe (see SlotLock.removeOrphanLogical).
-                    SlotLock.removeOrphanLogical(sfDir);
-                } catch (Throwable ignored) {
+                if (reclaimLogicalSlotLock) {
+                    try {
+                        // The logical slot lock lives OUTSIDE the slot dir (in the
+                        // shared .slot-locks dir) so it survives a slot rename; the
+                        // fully-drained retirement that removes this slot's other
+                        // side-files must remove it too, or .slot-locks accumulates a
+                        // dead lock+pid pair per distinct slot name for the lifetime of
+                        // sf_dir. Holding the directory-local lock is NOT sufficient to
+                        // make this safe -- it says nothing about the LOGICAL lock, which
+                        // a caller higher in the same stack may hold. Only a caller that
+                        // knows it does not hold it may reclaim, hence the flag.
+                        SlotLock.removeOrphanLogical(sfDir);
+                    } catch (Throwable ignored) {
+                    }
                 }
             }
         } finally {
@@ -812,7 +832,13 @@ public int recoverySymbolNativeCapacity() {
 
     private RecoveredFrameAnalysis checkedRecoveryAnalysis(int baseline) {
         if (recoveredFrameAnalysis == null || recoveredFrameAnalysis.baseline() != baseline) {
-            throw new IllegalStateException("recovery symbol baseline mismatch [expected="
+            // UnreplayableSlotException, NOT IllegalStateException: Sender.build() routes
+            // on the type, and only this type reaches its quarantine handler. A raw
+            // IllegalStateException escapes build() instead, and because senderId is
+            // stable and a not-fully-drained slot is retained on close, every restart
+            // re-recovers the same slot and rethrows -- the application can never
+            // construct a Sender, so it cannot even BUFFER new rows.
+            throw new UnreplayableSlotException("recovery symbol baseline mismatch [expected="
                     + (recoveredFrameAnalysis == null ? "none" : recoveredFrameAnalysis.baseline())
                     + ", actual=" + baseline + ']');
         }

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

@@ -726,7 +726,7 @@ public CursorWebSocketSendLoop(WebSocketClient client, CursorSendEngine engine,
         // simply starts empty and grows from the frames themselves.
         PersistedSymbolDict pd = engine.getPersistedSymbolDict();
         int persistedPrefixLen = 0;
-        if (pd != null && pd.size() > 0) {
+        if (pd != null && pd.recoveredSize() > 0) {
             int len = pd.loadedEntriesLen();
             if (len > 0) {
                 // Seed by reference. The foreground loop takes ownership after
@@ -739,10 +739,12 @@ public CursorWebSocketSendLoop(WebSocketClient client, CursorSendEngine engine,
                 sentDictBytesCapacity = len;
                 sentDictBytesLen = len;
                 // Set the count only alongside the bytes so sentDictCount can
-                // never claim symbols the mirror does not hold. A recovered slot
-                // always has loadedEntriesLen > 0 when size > 0, so this is the
-                // same result -- it just makes the coupling explicit.
-                sentDictCount = pd.size();
+                // never claim symbols the mirror does not hold -- and take it from
+                // recoveredSize(), NOT the live size(). They differ: the recovery
+                // heal (QwpWebSocketSender.healPersistedDictionary) appends the
+                // frame-contributed suffix to the file before this loop is built, so
+                // size() would already have run past the loaded byte region.
+                sentDictCount = pd.recoveredSize();
             }
         }
         // ...and then from the surviving frames' own delta sections, exactly as the producer