Fix delta symbol-dict recovery and reconnect bugs

Close three ways the delta symbol-dictionary feature could lose or
corrupt data on the reconnect and store-and-forward recovery paths.

Run the torn-dictionary guard unconditionally. trySendOne gated the
guard on deltaDictEnabled, which CursorSendEngine reports false when a
recovered disk slot cannot open its persisted dictionary (fd
exhaustion, a read-only remount, ENOSPC). The recorded frames are still
delta frames, so replaying them against a fresh empty-dictionary server
null-padded the missing ids and silently corrupted the table. The guard
now decodes the delta start for every frame and fails terminally on a
gap regardless of the flag; only the sent-dictionary mirror stays gated.

Stop treating a catch-up frame as the head data frame. sendCatchUpChunk
advances nextWireSeq, but onClose's poison-strike gate and
handleServerRejection's pre-send gate read nextWireSeq > 0 as "a data
frame was sent". A transient non-orderly close or NACK after the catch-up
but before the first replay frame then charged a poison strike on a frame
that never left, and after a few flaps escalated a transient outage to a
PROTOCOL_VIOLATION terminal that quarantines an orphan drainer. A new
dataFrameSentThisConnection flag, set only after a real ring frame sends,
now gates both decisions, so the drainer keeps retrying as Invariant B
requires.

Bound the commit message's dictionary delta to the sent watermark.
sendCommitMessage skips the write-ahead persist yet encoded a delta up to
currentBatchMaxSymbolId, so a symbol left in the batch by a cancelled row
(cancelRow rolls back neither currentBatchMaxSymbolId nor the global
registration) rode out on the commit frame without being persisted. A
recovered slot then under-seeded the producer against the surviving frame
and misattributed the reused id. The commit now caps the delta at
sentMaxSymbolId in delta mode, giving an empty delta.

Each fix carries a regression test proven to fail when the fix is
reverted: a directory-shadowed .symbol-dict (guard), a close after only
the catch-up (poison gate), and a cancelled-row symbol on a transactional
commit (delta bound).

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

@@ -3623,10 +3623,23 @@ private void sendCommitMessage() {
             LOG.debug("Sending commit message for deferred batch");
         }
         encoder.setDeferCommit(false);
-        // A commit carries no rows and no new symbols; in delta mode its empty
-        // delta simply starts at the server's current dictionary size.
+        // A commit carries no rows, and it must also carry NO new symbols. Unlike
+        // the flush paths, sendCommitMessage does NOT write-ahead-persist the
+        // dictionary, so shipping a symbol here would put an id on the wire that a
+        // recovered slot cannot rebuild from the persisted .symbol-dict, diverging
+        // the producer dictionary from the surviving frames and silently
+        // misattributing reused ids after a crash. currentBatchMaxSymbolId can sit
+        // ABOVE sentMaxSymbolId (e.g. a cancelled row: cancelRow does not roll back
+        // currentBatchMaxSymbolId or unregister the symbol), so bound the delta at
+        // what has already been sent -- and therefore already persisted. In delta
+        // mode pass sentMaxSymbolId, yielding an empty delta
+        // [sentMaxSymbolId+1 .. sentMaxSymbolId]; in full-dict mode keep
+        // currentBatchMaxSymbolId so the frame stays self-sufficient. Any symbol a
+        // cancelled row leaked is picked up (and persisted) by the next real flush,
+        // whose persistNewSymbolsBeforePublish resumes from pd.size().
+        int commitBatchMaxId = deltaDictEnabled ? sentMaxSymbolId : currentBatchMaxSymbolId;
         encoder.beginMessage(0, globalSymbolDictionary,
-                symbolDeltaBaseline(), currentBatchMaxSymbolId);
+                symbolDeltaBaseline(), commitBatchMaxId);
         int messageSize = encoder.finishMessage();
         QwpBufferWriter buffer = encoder.getBuffer();
         ensureActiveBufferReady();

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

@@ -203,6 +203,18 @@ 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;
+    // 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;
@@ -1980,6 +1992,11 @@ private void swapClient(WebSocketClient newClient) {
      * first real connection via swapClient.
      */
     private void setWireBaselineWithCatchUp(long replayStart) {
+        // Fresh connection: no data frame has been sent on it yet. Reset before the
+        // catch-up (which sends only dictionary frames) so onClose /
+        // handleServerRejection can tell "only the catch-up went out" from "the
+        // head data frame went out".
+        dataFrameSentThisConnection = false;
         if (client != null && deltaDictEnabled && sentDictCount > 0) {
             this.nextWireSeq = 0L;
             // The catch-up may span several frames when the dictionary exceeds the
@@ -2338,37 +2355,44 @@ private boolean trySendOne() {
             return false; // payload not fully published yet
         }
         long frameAddr = base + sendOffset + MmapSegment.FRAME_HEADER_SIZE;
-        // -1 unless this is a delta frame; the guard decodes it once here and
-        // accumulateSentDict reuses it post-send, so the delta header is parsed
-        // once per frame rather than twice.
-        int deltaStart = -1;
-        if (deltaDictEnabled) {
-            // Torn-dictionary guard. In normal operation a delta frame's start id
-            // never exceeds the dictionary coverage established so far (replayed
-            // frames overlap the catch-up dict; fresh frames extend it
-            // contiguously). A gap here means the persisted dictionary was torn --
-            // almost always by a host/power crash, which leaves segment frames on
-            // disk but loses recently-written dictionary entries (SF, like the rest
-            // of the store, is process-crash durable but not host-crash durable).
-            // Sending the frame would corrupt the table (the server would null-pad
-            // the missing ids), so fail terminally instead; the unreplayable data
-            // must be resent.
-            deltaStart = frameDeltaStart(frameAddr, payloadLen);
-            if (deltaStart > sentDictCount) {
-                recordFatal(new LineSenderException(
-                        "recovered store-and-forward symbol dictionary is incomplete (likely a host crash): "
-                                + "frame delta start " + deltaStart + " exceeds recovered dictionary size "
-                                + sentDictCount + "; cannot replay without corrupting data -- resend required"));
-                return false;
-            }
+        // Torn-dictionary guard. Decode the delta start unconditionally (-1 for a
+        // non-delta frame); the guard MUST run even when deltaDictEnabled is false.
+        // A disk slot recovered with its persisted dictionary unavailable
+        // (PersistedSymbolDict.open() returned null -- fd exhaustion, a read-only
+        // remount, ENOSPC) reports deltaDictEnabled=false, yet its recorded frames
+        // are still DELTA frames (deltaStart > 0). Replaying those against a fresh
+        // empty-dictionary server would null-pad the missing ids and SILENTLY
+        // corrupt the table -- precisely what this guard exists to prevent -- so it
+        // cannot be gated on the very flag that goes false in that failure mode. In
+        // normal operation a delta frame's start id never exceeds the dictionary
+        // coverage established so far (replayed frames overlap the catch-up dict;
+        // fresh frames extend it contiguously), so a gap here means the recovered
+        // dictionary is incomplete (a host/power crash that lost recently-written
+        // entries, SF being process-crash but not host-crash durable). Fail
+        // terminally; the unreplayable data must be resent. Full-dict / fallback
+        // frames carry deltaStart=0 with sentDictCount=0, so 0 > 0 never
+        // false-positives; only the sent-dictionary mirror below stays gated on
+        // deltaDictEnabled.
+        int deltaStart = frameDeltaStart(frameAddr, payloadLen);
+        if (deltaStart > sentDictCount) {
+            recordFatal(new LineSenderException(
+                    "recovered store-and-forward symbol dictionary is incomplete (likely a host crash): "
+                            + "frame delta start " + deltaStart + " exceeds recovered dictionary size "
+                            + sentDictCount + "; cannot replay without corrupting data -- resend required"));
+            return false;
         }
         try {
             client.sendBinary(frameAddr, payloadLen);
         } catch (Throwable t) {
             fail(t);
             return false;
         }
-        if (deltaStart >= 0) {
+        // A real ring frame (data or commit) has now gone out on this connection,
+        // as opposed to only the dictionary catch-up. onClose / handleServerRejection
+        // key their poison-strike vs pre-send decision off this, not off nextWireSeq
+        // (which the catch-up advances).
+        dataFrameSentThisConnection = true;
+        if (deltaDictEnabled && deltaStart >= 0) {
             // Mirror the symbols this frame introduced so a later reconnect can
             // rebuild the whole dictionary. Idempotent on replay: a frame whose
             // delta we already hold advances nothing.
@@ -2619,7 +2643,7 @@ public void onClose(int code, String reason) {
                     || code == WebSocketCloseCode.GOING_AWAY;
             LineSenderException cause = new LineSenderException(
                     "WebSocket closed by server: code=" + code + " reason=" + reason);
-            if (!orderly && nextWireSeq > 0) {
+            if (!orderly && dataFrameSentThisConnection) {
                 if (recordHeadRejectionStrike(Math.max(engine.ackedFsn(), highestOkFsn) + 1L)) {
                     haltOnPoisonedFrame("ws-close[" + code + ' '
                                     + WebSocketCloseCode.describe(code) + "]: " + reason,
@@ -2726,17 +2750,21 @@ private void handleServerRejection(long wireSeq) {
             // value is only used to attribute an FSN to the error report --
             // a rejection never advances the watermark.
             long highestSent = nextWireSeq - 1L;
-            if (highestSent < 0L) {
-                // Pre-send rejection: server emitted an error frame before
-                // we sent anything on this connection (typical after a
-                // fresh swapClient — auth failure, server-initiated halt,
-                // etc.). The server-named wireSeq does not correspond to
-                // any frame we sent, so clamping it to 0 and acknowledging
-                // fsnAtZero would silently advance ackedFsn past a real
-                // unsent batch (fsnAtZero == ackedFsn + 1 right after a
-                // swap). Skip the watermark advance entirely; still surface
-                // the error so the user's handler sees it and HALT errors
-                // remain producer-observable.
+            if (!dataFrameSentThisConnection) {
+                // Pre-send rejection: the server emitted an error frame before we
+                // sent any DATA frame on this connection (typical after a fresh
+                // swapClient -- auth failure, server-initiated halt, or a rejection
+                // of the dictionary catch-up itself). nextWireSeq may be > 0 here
+                // because the catch-up consumed wire sequences, so this keys off
+                // dataFrameSentThisConnection, not highestSent >= 0 -- otherwise a
+                // transient NACK of a catch-up frame would take the post-send
+                // poison-strike path and could escalate a transient outage to a
+                // terminal. The server-named wireSeq does not correspond to any
+                // data frame we sent, so clamping it to 0 and acknowledging
+                // fsnAtZero would silently advance ackedFsn past a real unsent
+                // batch. Skip the watermark advance entirely; still surface the
+                // error so the user's handler sees it and HALT errors remain
+                // producer-observable.
                 handlePreSendRejection(wireSeq, status, category, policy);
                 return;
             }

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

@@ -227,6 +227,139 @@ public void testTornDictionaryFailsCleanlyInsteadOfCorrupting() throws Exception
         });
     }
 
+    @Test
+    public void testUnopenablePersistedDictStillGuardsAgainstReplayingDeltaFrames() throws Exception {
+        // C1 regression: when a recovered disk slot's persisted dictionary cannot be
+        // OPENED (fd exhaustion, a read-only remount, ENOSPC -- simulated here by a
+        // .symbol-dict that is a DIRECTORY, so both openRW and openCleanRW fail),
+        // CursorSendEngine.isDeltaDictEnabled() returns false. The recorded frames
+        // are still DELTA frames, and replaying them against a fresh
+        // empty-dictionary server would null-pad the missing ids and SILENTLY
+        // corrupt the table. The torn-dictionary guard must fire regardless of
+        // deltaDictEnabled -- pre-fix it was gated on that very flag, so the
+        // corrupting frame sailed through unguarded. Unlike
+        // testTornDictionaryFailsCleanlyInsteadOfCorrupting (dict present but empty,
+        // deltaDictEnabled=true), here the dict is UNOPENABLE (deltaDictEnabled=false).
+        assertMemoryLeak(() -> {
+            // Phase 1: each row introduces a new symbol => frame i carries deltaStart=i.
+            try (TestWebSocketServer silent = new TestWebSocketServer(new SilentHandler())) {
+                int port = silent.getPort();
+                silent.start();
+                Assert.assertTrue(silent.awaitStart(5, TimeUnit.SECONDS));
+                String cfg = "ws::addr=localhost:" + port + ";sf_dir=" + sfDir
+                        + ";close_flush_timeout_millis=0;";
+                try (Sender s1 = Sender.fromConfig(cfg)) {
+                    for (int i = 0; i < 6; i++) {
+                        s1.table("m").symbol("s", "sym-" + i).longColumn("v", i).atNow();
+                        s1.flush();
+                    }
+                }
+            }
+
+            // Make the persisted dictionary UNOPENABLE: replace the .symbol-dict file
+            // with a directory of the same name so PersistedSymbolDict.open() returns
+            // null (both openRW and openCleanRW fail) and the engine reports
+            // deltaDictEnabled=false. Stamp the watermark at FSN 2 so replay starts
+            // at FSN 3 -- a frame whose delta starts at id 3, with ids 0..2 living
+            // only in the now-unreadable dictionary.
+            java.nio.file.Path slot = Paths.get(sfDir, "default");
+            java.nio.file.Path dict = slot.resolve(".symbol-dict");
+            java.nio.file.Files.delete(dict);
+            java.nio.file.Files.createDirectory(dict);
+            writeAckWatermark(slot.resolve(".ack-watermark"), 2);
+
+            // Phase 2: recover against a fresh counting server. The guard must fire
+            // (frame deltaStart 3 > recovered dictionary size 0) and fail terminally
+            // rather than send a gapped frame that would corrupt the table.
+            CountingHandler handler = new CountingHandler();
+            try (TestWebSocketServer good = new TestWebSocketServer(handler)) {
+                int port = good.getPort();
+                good.start();
+                Assert.assertTrue(good.awaitStart(5, TimeUnit.SECONDS));
+
+                String cfg = "ws::addr=localhost:" + port + ";sf_dir=" + sfDir + ";";
+                LineSenderException terminal = null;
+                Sender s2 = Sender.fromConfig(cfg);
+                try {
+                    long deadline = System.currentTimeMillis() + 10_000;
+                    while (System.currentTimeMillis() < deadline && terminal == null) {
+                        try {
+                            s2.flush();
+                            Thread.sleep(20);
+                        } catch (LineSenderException e) {
+                            terminal = e;
+                        }
+                    }
+                } finally {
+                    try {
+                        s2.close();
+                    } catch (LineSenderException e) {
+                        if (terminal == null) {
+                            terminal = e;
+                        }
+                    }
+                }
+                Assert.assertEquals("no delta frame may be replayed when the persisted dictionary is unopenable",
+                        0, handler.frames.get());
+                Assert.assertNotNull("an unopenable dictionary must surface a terminal error", terminal);
+                Assert.assertTrue(terminal.getMessage(),
+                        terminal.getMessage().contains("symbol dictionary is incomplete"));
+            }
+        });
+    }
+
+    @Test
+    public void testCommitMessageDoesNotShipUnpersistedLeakedSymbol() throws Exception {
+        // C3 regression: sendCommitMessage does NOT write-ahead-persist the
+        // dictionary, so its frame must carry NO new symbol. A symbol left in the
+        // batch by a cancelled row -- cancelRow rolls back neither
+        // currentBatchMaxSymbolId nor the global-dictionary registration -- must not
+        // ride out on the commit frame: doing so puts an id on the wire that a
+        // recovered slot cannot rebuild from .symbol-dict, diverging the producer
+        // dictionary from the surviving frames and silently misattributing reused
+        // ids after a crash. The commit's delta must be bounded by sentMaxSymbolId
+        // (empty here), not currentBatchMaxSymbolId. Memory mode suffices to observe
+        // the wire behaviour; close() drains every frame to the server first.
+        assertMemoryLeak(() -> {
+            DictReconstructingHandler handler = new DictReconstructingHandler();
+            try (TestWebSocketServer server = new TestWebSocketServer(handler)) {
+                int port = server.getPort();
+                server.start();
+                Assert.assertTrue(server.awaitStart(5, TimeUnit.SECONDS));
+
+                // Transactional + autoFlushRows=1: every completed row auto-flushes
+                // as a DEFERRED batch (setting hasDeferredMessages); an explicit
+                // flush() then emits the commit message.
+                Sender sender = Sender.builder("ws::addr=localhost:" + port + ";")
+                        .transactional(true)
+                        .autoFlushRows(1)
+                        .build();
+                try {
+                    // Row 1 registers "a"@0 and auto-flushes it deferred.
+                    sender.table("m").symbol("s", "a").longColumn("v", 1L).atNow();
+                    // Register "b"@1 on a row that is then cancelled: "b" stays in
+                    // the global dictionary and currentBatchMaxSymbolId advances to
+                    // 1, but nothing persists or sends it.
+                    sender.table("m").symbol("s", "b");
+                    sender.cancelRow();
+                    // Commit the deferred batch. The commit frame must carry an
+                    // EMPTY delta -- NOT "b"@1.
+                    sender.flush();
+                } finally {
+                    sender.close(); // drains every frame (incl. the commit) to the server
+                }
+
+                // The server's reconstructed dictionary must hold ONLY "a". Pre-fix
+                // the commit shipped "b"@1, so the server saw a second symbol.
+                List<String> dict = handler.dictSnapshot();
+                Assert.assertEquals("commit frame must not ship the cancelled row's leaked symbol "
+                                + "(recovery would then diverge from the persisted dictionary): " + dict,
+                        1, dict.size());
+                Assert.assertEquals("a", dict.get(0));
+            }
+        });
+    }
+
     @Test
     public void testFailedPublishDoesNotDuplicatePersistedSymbols() throws Exception {
         // Regression: persistNewSymbolsBeforePublish is a write-ahead -- it runs