Harden delta symbol-dict recovery and NACK gating

Two fixes surfaced by review of the delta symbol-dictionary change.

Recover oversized persisted symbols (C1). PersistedSymbolDict.openExisting
rejected any entry larger than a fixed 1 MB ceiling as "oversized" and
truncated the dictionary at that id, but the write path
(appendSymbol/appendSymbols) caps nothing. A symbol value over 1 MB
therefore persisted fine, yet a normal process-crash recovery dropped it
and every higher id; the send loop's replay guard then fired a spurious
"host crash / resend required" terminal -- a store-and-forward
process-crash-durability violation on a reachable input (nothing caps
symbol value length). Drop the fixed per-entry ceiling and bound entries
only by the file length, which is the actual corruption guard and was
already present. Keep the length in a long so a corrupt multi-gigabyte
varint cannot wrap an int back under the check. The write and read paths
now agree, so a legitimately persisted large symbol recovers.

Guard the catch-up NACK pre-send gate (C2). handleServerRejection keys
its pre-send branch off dataFrameSentThisConnection, because the symbol
catch-up advances nextWireSeq without sending a data frame. That branch
had no regression test: every existing NACK test sets both flags
together, so reverting the gate to the old nextWireSeq-based predicate
left the suite green. Add the server-NACK twin of
testNonOrderlyCloseAfterOnlyCatchUpDoesNotStrike -- a WRITE_ERROR NACK of
a catch-up frame must take the pre-send path (surface and recycle, no
strike), not escalate a transient outage to a producer-fatal
PROTOCOL_VIOLATION terminal.

Both regression tests fail with their production line reverted and pass
with it.

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

Author: glasstiger

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

@@ -89,9 +89,6 @@ public final class PersistedSymbolDict implements QuietCloseable {
     static final int FILE_MAGIC = 0x31445953; // 'SYD1' little-endian
     static final int HEADER_SIZE = 8;
     static final byte VERSION = 1;
-    // Guards against a hostile/corrupt varint length driving a huge allocation
-    // or a runaway parse. Symbols are short; this is a generous ceiling.
-    private static final int MAX_ENTRY_LEN = 1 << 20;
     private static final Logger LOG = LoggerFactory.getLogger(PersistedSymbolDict.class);
     private final int fd;
     private long appendOffset;
@@ -337,12 +334,20 @@ private static PersistedSymbolDict openExisting(String filePath, long fileLen) {
                 if (vr == null) {
                     break; // torn length varint
                 }
-                int entryLen = (int) vr[0];
+                long entryLen = vr[0];
                 int next = (int) vr[1];
-                if (entryLen < 0 || entryLen > MAX_ENTRY_LEN || (long) next + entryLen > len) {
-                    break; // torn/oversized entry -- self-healing tail
+                // The file length is the sole sanity bound: a length that runs past
+                // the file is a torn/incomplete trailing entry and stops the parse
+                // (self-healing tail). There is deliberately NO fixed per-entry
+                // ceiling -- the write path (appendSymbol/appendSymbols) applies none
+                // either, so a legitimately persisted large symbol must recover here,
+                // not be truncated and then trip the send loop's "resend required"
+                // guard on a normal process-crash recovery. entryLen stays a long so a
+                // corrupt multi-gigabyte length cannot wrap an int back under the check.
+                if ((long) next + entryLen > len) {
+                    break; // torn/incomplete trailing entry -- self-healing tail
                 }
-                pos = next + entryLen;
+                pos = next + (int) entryLen;
                 count++;
             }
             int entriesLen = pos - HEADER_SIZE;

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

@@ -279,6 +279,42 @@ public void testNonOrderlyCloseAfterOnlyCatchUpDoesNotStrike() throws Exception
         });
     }
 
+    @Test
+    public void testNonOrderlyRejectionAfterOnlyCatchUpDoesNotStrike() throws Exception {
+        // C2 regression: the server-NACK twin of
+        // testNonOrderlyCloseAfterOnlyCatchUpDoesNotStrike. handleServerRejection's
+        // pre-send gate keys off dataFrameSentThisConnection, NOT nextWireSeq -- the
+        // dictionary catch-up advances nextWireSeq WITHOUT sending a data frame. A
+        // server NACK of a catch-up frame (nextWireSeq>0, no data frame sent) must
+        // take the pre-send path (surface + recycle, no strike), not the post-send
+        // poison-strike path. Pre-fix (gate on highestSent >= 0, i.e. nextWireSeq>0)
+        // each catch-up NACK strikes the never-sent head frame; MAX_REJECTIONS such
+        // strikes escalate a TRANSIENT outage to a producer-fatal PROTOCOL_VIOLATION
+        // terminal -- exactly what store-and-forward's retry-forever contract forbids.
+        // WRITE_ERROR (RETRIABLE) is the discriminating category: it DOES accrue
+        // strikes on the post-send path (see testNackRecycleIsPacedAgainstHealthyServer),
+        // so a regressed gate escalates here and checkError() throws.
+        TestUtils.assertMemoryLeak(() -> {
+            List<WebSocketClient> clients = new ArrayList<>();
+            try (CursorSendEngine engine = newEngine()) {
+                appendFrames(engine, 2);
+                CursorWebSocketSendLoop loop = newDurableLoop(engine, clients);
+                for (int i = 0; i < MAX_REJECTIONS + 2; i++) {
+                    // Model the catch-up: nextWireSeq advanced, but NO data frame
+                    // sent. The pre-send recycle (swapClient) resets nextWireSeq, so
+                    // re-apply before each NACK (mirrors the onClose twin).
+                    setCatchUpWireSeqOnly(loop, 2);
+                    deliverRetriableNack(loop, 1, "disk full");
+                }
+                // No strike was ever charged, so nothing escalated: the loop stays
+                // retriable and the producer-facing error latch is clear.
+                loop.checkError();
+            } finally {
+                closeAll(clients);
+            }
+        });
+    }
+
     @Test
     public void testNackRecycleIsPacedAgainstHealthyServer() throws Exception {
         // A reachable, healthy server that NACKs the head frame (RETRIABLE)

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

@@ -27,6 +27,7 @@
 import io.questdb.client.cutlass.qwp.client.GlobalSymbolDictionary;
 import io.questdb.client.cutlass.qwp.client.sf.cursor.PersistedSymbolDict;
 import io.questdb.client.std.ObjList;
+import io.questdb.client.test.tools.TestUtils;
 import org.junit.Assert;
 import org.junit.Test;
 
@@ -198,6 +199,50 @@ public void testEmptySymbolRoundTrips() throws Exception {
         });
     }
 
+    @Test
+    public void testLargeSymbolRoundTripsAcrossReopen() throws Exception {
+        // C1 regression: the write path caps nothing, so a symbol larger than the
+        // old fixed 1 MB read ceiling must still recover intact. Before the fix,
+        // appendSymbol wrote the oversized entry but openExisting rejected it as
+        // "oversized", truncated the dictionary at that id (dropping it and every
+        // higher id), and a normal process-crash recovery then hard-failed with a
+        // spurious "host crash / resend required" terminal -- defeating store-and-
+        // forward's process-crash durability for large symbols. The file length is
+        // now the only bound, so the write and read paths agree.
+        assertMemoryLeak(() -> {
+            Path dir = Files.createTempDirectory("qwp-symdict");
+            try {
+                // Just over the old 1 << 20 (1 MB) ceiling.
+                String big = TestUtils.repeat("x", (1 << 20) + 17);
+                PersistedSymbolDict d = PersistedSymbolDict.open(dir.toString());
+                try {
+                    d.appendSymbol("before");
+                    d.appendSymbol(big);
+                    d.appendSymbol("after");
+                    Assert.assertEquals(3, d.size());
+                } finally {
+                    d.close();
+                }
+
+                // Recovery must load ALL three; pre-fix the reopen truncated at the
+                // big entry and came back with size 1 (only "before" survived).
+                PersistedSymbolDict re = PersistedSymbolDict.open(dir.toString());
+                try {
+                    Assert.assertEquals("large entry must survive recovery, not be truncated",
+                            3, re.size());
+                    ObjList<String> s = re.readLoadedSymbols();
+                    Assert.assertEquals("before", s.getQuick(0));
+                    Assert.assertEquals(big, s.getQuick(1));
+                    Assert.assertEquals("after", s.getQuick(2));
+                } finally {
+                    re.close();
+                }
+            } finally {
+                rmDir(dir);
+            }
+        });
+    }
+
     @Test
     public void testRemoveOrphanDeletesFile() throws Exception {
         assertMemoryLeak(() -> {