Harden delta symbol-dict send and persist paths

Four follow-ups from the delta symbol-dictionary review.

Catch-up NACK no longer launders the poison detector. After a reconnect
the loop ships the head data frame before tryReceiveAcks reads the
server's NACK of a dictionary catch-up frame, so handleServerRejection
saw dataFrameSentThisConnection=true and attributed the NACK a wire seq
below the replay head -- fsn = fsnAtZero+cappedSeq, negative when
replayStart < catchUpFrames. recordHeadRejectionStrike then set poisonFsn
to that value (often -1, the "no suspect" sentinel), wiping a genuine
in-progress poison run and reporting a bogus fsn. Route an
fsn <= ackedFsn rejection to the pre-send path (surface + recycle, no
strike, no watermark advance), symmetric with the success path's
engine.acknowledge() no-op below ackedFsn. A real replayed data frame is
at fsn > ackedFsn, so it is never caught. Regression test:
testPostSendCatchUpNackDoesNotStrikeOrLaunderPoisonState.

Persist short-write now surfaces as LineSenderException. A disk-full
during the write-ahead persist threw a raw IllegalStateException that
escaped flush() unwrapped, unlike every other flush-path failure (e.g.
the cursor append in sealAndSwapBuffer). Wrap it so a caller catching
LineSenderException around flush() sees a persist disk-full too; a JVM
Error still propagates.

PersistedSymbolDict close() and the append methods are synchronized.
close() is callable from any thread (a shutdown hook); without mutual
exclusion a close racing an in-flight append could free the scratch
buffer or close the fd mid-write and let the write land on a descriptor
the OS has reused for another file (silent cross-file corruption).

The appendSymbols re-encode branch now has a regression test. After a
failed publish leaves the durable dictionary size ahead of the wire
baseline, a later flush introducing a new symbol re-encodes only the
[pd.size() .. currentBatchMaxSymbolId] suffix; keying that off pd.size()
(not sentMaxSymbolId+1) keeps it idempotent. Regression test:
testFailedPublishThenNewSymbolPersistsSuffixWithoutDuplicating.

Both new tests fail with their production hunk reverted.

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

@@ -3707,14 +3707,30 @@ private void persistNewSymbolsBeforePublish() {
         // failed publish the durable size has run ahead of the wire baseline, so
         // the frame's delta covers MORE than remains to persist; then re-encode
         // just the [from .. currentBatchMaxSymbolId] suffix.
-        if (from == sentMaxSymbolId + 1) {
-            QwpBufferWriter buffer = encoder.getBuffer();
-            pd.appendRawEntries(
-                    buffer.getBufferPtr() + encoder.getDeltaEntriesStart(),
-                    encoder.getDeltaEntriesLen(),
-                    currentBatchMaxSymbolId - from + 1);
-        } else {
-            pd.appendSymbols(globalSymbolDictionary, from, currentBatchMaxSymbolId);
+        try {
+            if (from == sentMaxSymbolId + 1) {
+                QwpBufferWriter buffer = encoder.getBuffer();
+                pd.appendRawEntries(
+                        buffer.getBufferPtr() + encoder.getDeltaEntriesStart(),
+                        encoder.getDeltaEntriesLen(),
+                        currentBatchMaxSymbolId - from + 1);
+            } else {
+                pd.appendSymbols(globalSymbolDictionary, from, currentBatchMaxSymbolId);
+            }
+        } catch (Throwable t) {
+            // A short write (disk full / quota) to the persisted dictionary throws
+            // a low-level IllegalStateException. Surface it as a LineSenderException
+            // -- like every other flush-path failure, e.g. the cursor append in
+            // sealAndSwapBuffer -- so a caller catching LineSenderException around
+            // flush() also catches a disk-full during the write-ahead persist. The
+            // persist ran before publish and pd.size() did not advance on the short
+            // write, so the still-buffered rows re-persist the same range
+            // idempotently on retry. A JVM Error is never a persist failure; let it
+            // propagate.
+            if (t instanceof Error) {
+                throw (Error) t;
+            }
+            throw new LineSenderException("failed to persist symbol dictionary before publish", t);
         }
     }
 

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

@@ -2811,6 +2811,25 @@ private void handleServerRejection(long wireSeq) {
                         wireSeq, highestSent);
             }
             long fsn = fsnAtZero + cappedSeq;
+            if (fsn <= engine.ackedFsn()) {
+                // The clamped wire seq maps at or below the replay head, so this
+                // NACK is for a dictionary catch-up frame -- which occupies the
+                // already-acked wire sequences below replayStart -- not a data frame
+                // this connection sent. (dataFrameSentThisConnection can be true here
+                // because trySendOne ships the head data frame before tryReceiveAcks
+                // reads the catch-up's NACK in the same loop iteration.) Attributing
+                // it a data FSN would key recordHeadRejectionStrike() off a
+                // below-baseline FSN -- negative when replayStart < catchUpFrames --
+                // colliding with the poisonFsn == -1 "no suspect" sentinel and
+                // laundering a genuine poison run, and would report a bogus FSN.
+                // Treat it exactly like a pre-send rejection: surface + recycle, no
+                // poison strike, no watermark advance. Symmetric with the success
+                // path, where engine.acknowledge() no-ops at or below ackedFsn. A
+                // real replayed data frame is at fsn > ackedFsn, so it is never
+                // caught here.
+                handlePreSendRejection(wireSeq, status, category, policy);
+                return;
+            }
             // Best-effort table attribution: the parser populates
             // response.tableNames on error frames the same way it does on
             // STATUS_OK. If exactly one table was named, surface it; if

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

@@ -83,9 +83,14 @@
  * A torn trailing entry from a crash mid-append is self-healing: {@link #open}
  * stops parsing at the first incomplete entry and the next append overwrites it.
  * <p>
- * <b>Lifecycle:</b> single-writer (the producer / user thread). Read once at
- * {@link #open} to seed in-memory state on recovery or orphan-drain. Owner
- * (the engine) closes it. Not thread-safe for concurrent writers.
+ * <b>Lifecycle:</b> single-writer (the producer / user thread) for appends. Read
+ * once at {@link #open} to seed in-memory state on recovery or orphan-drain. The
+ * owner (the engine) closes it, and {@code close()} is callable from any thread
+ * (a shutdown hook, test cleanup). {@code close()} and the append methods are
+ * therefore {@code synchronized}: without that, a close racing an in-flight append
+ * could free the scratch buffer or close the fd mid-write and let the write land
+ * on a descriptor the OS has reused for another file (silent cross-file
+ * corruption). Not thread-safe for concurrent writers.
  */
 public final class PersistedSymbolDict implements QuietCloseable {
 
@@ -188,7 +193,7 @@ public static void removeOrphan(String slotDir) {
      * appendOffset}, so a retry keyed off {@link #size()} re-persists the same range
      * at the same offset. No-op when the range is empty or the dictionary is closed.
      */
-    public void appendRawEntries(long addr, int len, int count) {
+    public synchronized void appendRawEntries(long addr, int len, int count) {
         if (closed || count <= 0 || len <= 0) {
             return;
         }
@@ -208,7 +213,7 @@ public void appendRawEntries(long addr, int len, int count) {
      * (see the class-level durability note). Assigns the next dense id implicitly
      * (the entry's position).
      */
-    public void appendSymbol(CharSequence symbol) {
+    public synchronized void appendSymbol(CharSequence symbol) {
         if (closed) {
             return;
         }
@@ -244,7 +249,7 @@ public void appendSymbol(CharSequence symbol) {
      * range and overwrites at the same offset. No-op when the range is empty or
      * the dictionary is closed.
      */
-    public void appendSymbols(GlobalSymbolDictionary dict, int from, int to) {
+    public synchronized void appendSymbols(GlobalSymbolDictionary dict, int from, int to) {
         if (closed || to < from) {
             return;
         }
@@ -270,7 +275,7 @@ public void appendSymbols(GlobalSymbolDictionary dict, int from, int to) {
     }
 
     @Override
-    public void close() {
+    public synchronized void close() {
         if (closed) {
             return;
         }

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

@@ -480,6 +480,70 @@ public void testFailedPublishDoesNotDuplicatePersistedSymbols() throws Exception
         });
     }
 
+    @Test
+    public void testFailedPublishThenNewSymbolPersistsSuffixWithoutDuplicating() throws Exception {
+        // Regression for the re-encode (else) branch of persistNewSymbolsBeforePublish.
+        // After a failed publish leaves the durable dictionary size ahead of the wire
+        // baseline (pd.size() > sentMaxSymbolId+1), a later flush that introduces a NEW
+        // symbol cannot reuse the frame's already-encoded fast-path bytes -- it
+        // re-encodes just the [pd.size() .. currentBatchMaxSymbolId] suffix via
+        // appendSymbols. Keying that off pd.size() (not sentMaxSymbolId+1) keeps it
+        // idempotent: the already-persisted prefix is NOT re-appended.
+        assertMemoryLeak(() -> {
+            try (TestWebSocketServer silent = new TestWebSocketServer(new SilentHandler())) {
+                int port = silent.getPort();
+                silent.start();
+                Assert.assertTrue(silent.awaitStart(5, TimeUnit.SECONDS));
+                // Small segment: any frame carrying the padded row fails to publish with
+                // PAYLOAD_TOO_LARGE deterministically (no backpressure timing).
+                String cfg = "ws::addr=localhost:" + port + ";sf_dir=" + sfDir + ";sf_max_bytes=1024;";
+                String pad = TestUtils.repeat("x", 2000); // frame >> 1024-byte segment
+                Sender sender = Sender.fromConfig(cfg);
+                try {
+                    // Flush 1: s0 is persisted (write-ahead) before the oversized frame
+                    // fails to publish. pd.size()=1, sentMaxSymbolId stays -1.
+                    sender.table("m").symbol("s", "s0").stringColumn("p", pad).longColumn("v", 1L).atNow();
+                    try {
+                        sender.flush();
+                        Assert.fail("oversized frame must fail to publish");
+                    } catch (LineSenderException expected) {
+                    }
+
+                    // Add a NEW symbol s1 (id 1). The failed s0 row is still buffered, so
+                    // the batch is {s0, s1} and the durable size (1) has run ahead of the
+                    // wire baseline (-1) -- the state that selects the appendSymbols branch.
+                    sender.table("m").symbol("s", "s1").stringColumn("p", pad).longColumn("v", 2L).atNow();
+                    try {
+                        sender.flush();
+                        Assert.fail("oversized frame must fail to publish");
+                    } catch (LineSenderException expected) {
+                    }
+
+                    // The else branch persisted ONLY s1 (the suffix). The dictionary holds
+                    // s0, s1 exactly once each. Pre-fix (appendSymbols from
+                    // sentMaxSymbolId+1) re-appended s0, giving size 3.
+                    PersistedSymbolDict pd = PersistedSymbolDict.open(Paths.get(sfDir, "default").toString());
+                    Assert.assertNotNull(pd);
+                    try {
+                        Assert.assertEquals("re-encode suffix must not duplicate the persisted prefix",
+                                2, pd.size());
+                        Assert.assertEquals("s0", pd.readLoadedSymbols().getQuick(0));
+                        Assert.assertEquals("s1", pd.readLoadedSymbols().getQuick(1));
+                    } finally {
+                        pd.close();
+                    }
+                } finally {
+                    try {
+                        sender.close();
+                    } catch (LineSenderException ignored) {
+                        // close() re-flushes the still-buffered oversized rows and fails
+                        // again (PAYLOAD_TOO_LARGE); expected, resources still released.
+                    }
+                }
+            }
+        });
+    }
+
     @Test
     public void testRecoveryAfterFailedPublishReplaysGapFree() throws Exception {
         // M3 end-to-end: chains a failed publish -> fresh-process recovery -> replay.

core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoopPoisonFrameTest.java

@@ -656,6 +656,46 @@ public void testPoisonDwellHoldsEscalationUntilWallClockWindowElapses() throws E
         });
     }
 
+    @Test
+    public void testPostSendCatchUpNackDoesNotStrikeOrLaunderPoisonState() throws Exception {
+        // Regression: after a reconnect the loop ships the head DATA frame
+        // (dataFrameSentThisConnection=true) BEFORE tryReceiveAcks reads the server's
+        // NACK of a dictionary CATCH-UP frame in the same loop iteration. That NACK
+        // names a wire seq below the replay head, so its fsn = fsnAtZero+cappedSeq is
+        // at or below ackedFsn -- negative when replayStart < catchUpFrames. It must
+        // NOT charge a poison strike: recordHeadRejectionStrike(fsn) would set
+        // poisonFsn to that value (the common shape yields -1, the "no suspect"
+        // sentinel), laundering any genuine in-progress poison run and reporting a
+        // bogus fsn. The fix routes an fsn <= ackedFsn rejection to the pre-send path
+        // (surface + recycle, no strike), symmetric with the success path's
+        // engine.acknowledge() no-op below ackedFsn.
+        TestUtils.assertMemoryLeak(() -> {
+            List<WebSocketClient> clients = new ArrayList<>();
+            try (CursorSendEngine engine = newEngine()) {
+                appendFrames(engine, 2);
+                CursorWebSocketSendLoop loop = newDurableLoop(engine, clients);
+                // Model: 2 catch-up frames (wire seq 0,1) + 1 data frame (wire seq 2)
+                // sent, nothing acked (ackedFsn = -1), so fsnAtZero = replayStart(0) -
+                // catchUpFrames(2) = -2. A NACK of catch-up wire seq 0 maps to fsn -2,
+                // strictly below the -1 sentinel so the assertion discriminates.
+                setPostCatchUpDataFrameBaseline(loop, -2L, 3L);
+                assertEquals("precondition: no poison suspect yet",
+                        -1L, getLongField(loop, "poisonFsn"));
+
+                deliverRetriableNack(loop, 0, "disk full");
+
+                // The catch-up NACK was surfaced + recycled but charged NO strike, so
+                // the sentinel is intact. Pre-fix it was set to -2, laundering the
+                // poison detector's state.
+                assertEquals("catch-up NACK must not set/launder the poison sentinel",
+                        -1L, getLongField(loop, "poisonFsn"));
+                loop.checkError();
+            } finally {
+                closeAll(clients);
+            }
+        });
+    }
+
     @Test
     public void testPostSendNotWritableNackNeverEscalatesToPoisonTerminal() throws Exception {
         // RETRIABLE_OTHER (NOT_WRITABLE) is a node-state verdict, not a frame
@@ -1058,6 +1098,28 @@ private static void setCatchUpWireSeqOnly(CursorWebSocketSendLoop loop, long cou
         f.setLong(loop, count);
     }
 
+    // Models a fresh connection that has shipped the dictionary catch-up (advancing
+    // fsnAtZero below the replay head) AND the head data frame: sets fsnAtZero,
+    // nextWireSeq, and dataFrameSentThisConnection=true together.
+    private static void setPostCatchUpDataFrameBaseline(CursorWebSocketSendLoop loop,
+                                                        long fsnAtZero, long nextWireSeq) throws Exception {
+        Field z = CursorWebSocketSendLoop.class.getDeclaredField("fsnAtZero");
+        z.setAccessible(true);
+        z.setLong(loop, fsnAtZero);
+        Field w = CursorWebSocketSendLoop.class.getDeclaredField("nextWireSeq");
+        w.setAccessible(true);
+        w.setLong(loop, nextWireSeq);
+        Field d = CursorWebSocketSendLoop.class.getDeclaredField("dataFrameSentThisConnection");
+        d.setAccessible(true);
+        d.setBoolean(loop, true);
+    }
+
+    private static long getLongField(CursorWebSocketSendLoop loop, String name) throws Exception {
+        Field f = CursorWebSocketSendLoop.class.getDeclaredField(name);
+        f.setAccessible(true);
+        return f.getLong(loop);
+    }
+
     private static long[] txns(long... v) {
         return v;
     }