Pin the C3 and C4 fixes with regression tests

Each test was verified to fail with its production fix reverted, and the
revert was applied in place rather than via git checkout so no other
uncommitted work could be discarded.

testPersistFailureDegradesToFullDictInsteadOfKillingFlushForever arms the
full-disk facade before the first flush, so the very first ensureAppendMap
is refused -- a later append would sit inside the window already mapped
and never call allocate at all. The first flush must still throw, because
beginMessage has already baked a delta deltaStart into the staged frame,
but the second must succeed with deltaStart 0. Reverted, the retry fails
with "failed to persist symbol dictionary before publish".

testRecoveryBaselineMismatchIsQuarantinableNotAPermanentBrick recovers a
slot and asks addRecoveredSymbolsTo for a baseline the fold cannot match.
Reverted, it surfaces IllegalState "recovery symbol baseline mismatch
[expected=0, actual=9999]", which Sender.build() does not route to its
quarantine handler.

testUndecodableLoadedRegionIsQuarantinableNotAPermanentBrick hand-writes a
CRC-valid chunk declaring 3 entries inside a 2-byte region. Reverted, it
surfaces IllegalState "truncated loaded symbol dictionary entry". The
fixture also documents a gap worth closing separately: scanRecoveredChunks
verifies the chunk CRC and rejects entryCount <= 0, but never checks that
entryBytes actually holds entryCount well-formed entries, so it accepts
this chunk and only the decoder notices. Validating the triple during the
scan would end the trusted prefix there, as a CRC failure already does.

Full client suite: 2690 tests green.

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

Author: glasstiger

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

@@ -1761,6 +1761,80 @@ public void testRecoveryHealsThePersistedDictionaryBeforeAnyNewFrame() throws Ex
         });
     }
 
+    /**
+     * A dictionary that becomes unwritable mid-run must cost bandwidth, not the sender.
+     * <p>
+     * deltaDictEnabled was written once at setCursorEngine with no runtime disable, so
+     * the first persist failure killed flush() permanently -- every later flush re-threw.
+     * A full disk reaches exactly that state and is survivable: SF's segments are
+     * pre-allocated mmap files and stay writable while the growing .symbol-dict does not,
+     * and full self-sufficient frames need no side file at all (each carries the whole
+     * dictionary from id 0, which is what recovery replays anyway).
+     * <p>
+     * So the FIRST flush must still fail -- beginMessage has already baked a delta
+     * deltaStart into the staged frame, and publishing it would put ids on the ring the
+     * side-file cannot describe -- but the SECOND must succeed, in full-dict mode. Remove
+     * the disableDeltaDict call and the second flush throws exactly like the first.
+     */
+    @Test(timeout = 60_000L)
+    public void testPersistFailureDegradesToFullDictInsteadOfKillingFlushForever() throws Exception {
+        assertMemoryLeak(() -> {
+            DictReconstructingHandler handler = new DictReconstructingHandler();
+            try (TestWebSocketServer server = new TestWebSocketServer(handler)) {
+                int port = server.getPort();
+                server.start();
+                Assert.assertTrue(server.awaitStart(5, TimeUnit.SECONDS));
+
+                String slot = Paths.get(sfDir, "default").toString();
+                Assert.assertEquals(0, io.questdb.client.std.Files.mkdir(sfDir,
+                        io.questdb.client.std.Files.DIR_MODE_DEFAULT));
+                FullDiskDictFacade ff = new FullDiskDictFacade();
+                CursorSendEngine engine = new CursorSendEngine(
+                        slot, 4L * 1024 * 1024, CursorSendEngine.DEFAULT_APPEND_DEADLINE_NANOS,
+                        CursorSendEngine.DEFAULT_APPEND_DEADLINE_NANOS, ff);
+                Sender sender = QwpWebSocketSender.connect(
+                        "localhost", port, null, 0, 0, 0L, null, false, engine);
+                try {
+                    // Armed from the start, so the very first ensureAppendMap is refused --
+                    // a later append would sit inside the window already mapped and never
+                    // call allocate at all.
+                    ff.armed = true;
+                    sender.table("m").symbol("s", "a").longColumn("v", 1L).atNow();
+                    try {
+                        sender.flush();
+                        Assert.fail("the first flush must fail: its staged frame carries a delta "
+                                + "deltaStart the side-file cannot describe");
+                    } catch (LineSenderException expected) {
+                        Assert.assertTrue("expected the persist-failure message, got: "
+                                        + expected.getMessage(),
+                                expected.getMessage().contains("failed to persist symbol dictionary"));
+                    }
+
+                    // The rows are still buffered (the throw precedes every publish), so the
+                    // retry re-encodes them from id 0. Pre-fix this threw forever.
+                    sender.flush();
+
+                    long deadline = System.currentTimeMillis() + 5_000;
+                    while (System.currentTimeMillis() < deadline && handler.dataFrameCount() < 1) {
+                        Thread.sleep(20);
+                    }
+                    Assert.assertTrue("the degraded flush must actually reach the server",
+                            handler.dataFrameCount() >= 1);
+                    Assert.assertEquals("a degraded frame must be self-sufficient (deltaStart 0)",
+                            0, handler.lastDataDeltaStart());
+                    Assert.assertEquals("the full dictionary must ride along",
+                            java.util.Collections.singletonList("a"), handler.dictSnapshot());
+                } finally {
+                    try {
+                        sender.close();
+                    } catch (LineSenderException ignored) {
+                        // not what we assert
+                    }
+                }
+            }
+        });
+    }
+
     private static boolean hasSegmentFile(java.nio.file.Path dir) {
         java.io.File[] files = dir.toFile().listFiles();
         if (files != null) {

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

@@ -27,6 +27,8 @@
 import io.questdb.client.cutlass.line.LineSenderException;
 import io.questdb.client.cutlass.qwp.client.sf.cursor.AckWatermark;
 import io.questdb.client.cutlass.qwp.client.sf.cursor.CursorSendEngine;
+import io.questdb.client.cutlass.qwp.client.sf.cursor.UnreplayableSlotException;
+import io.questdb.client.cutlass.qwp.client.GlobalSymbolDictionary;
 import io.questdb.client.cutlass.qwp.client.sf.cursor.MmapSegment;
 import io.questdb.client.cutlass.qwp.client.sf.cursor.SegmentManager;
 import io.questdb.client.cutlass.qwp.client.sf.cursor.SlotLock;
@@ -57,6 +59,44 @@
 
 public class CursorSendEngineTest {
 
+    /**
+     * A recovery-seed baseline mismatch must be quarantinable, not a permanent brick.
+     * <p>
+     * checkedRecoveryAnalysis threw a raw IllegalStateException, and Sender.build() routes
+     * on the TYPE -- only UnreplayableSlotException reaches its quarantine handler. With a
+     * stable senderId and a not-fully-drained slot retained on close, every restart
+     * re-recovered the same slot and rethrew: the application could never construct a
+     * Sender, so it could not even BUFFER new rows. Revert the type and this goes red.
+     */
+    @Test(timeout = 10_000L)
+    public void testRecoveryBaselineMismatchIsQuarantinableNotAPermanentBrick() throws Exception {
+        TestUtils.assertMemoryLeak(() -> {
+            long buf = Unsafe.malloc(64, MemoryTag.NATIVE_DEFAULT);
+            try {
+                try (CursorSendEngine engine = new CursorSendEngine(tmpDir, 4096)) {
+                    engine.appendBlocking(buf, 64);
+                }
+            } finally {
+                Unsafe.free(buf, 64, MemoryTag.NATIVE_DEFAULT);
+            }
+            try (CursorSendEngine reopened = new CursorSendEngine(tmpDir, 4096)) {
+                assertTrue("scaffolding: the slot must have been recovered",
+                        reopened.wasRecoveredFromDisk());
+                try {
+                    // The fold was keyed to the persisted prefix size; ask for a baseline
+                    // that cannot match it.
+                    reopened.addRecoveredSymbolsTo(9_999, new GlobalSymbolDictionary());
+                    fail("a baseline mismatch must be reported");
+                } catch (UnreplayableSlotException expected) {
+                    assertTrue("expected the baseline-mismatch message, got: "
+                                    + expected.getMessage(),
+                            expected.getMessage().contains("recovery symbol baseline mismatch"));
+                }
+            }
+        });
+    }
+
+
     private String tmpDir;
 
     @Before

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

@@ -26,8 +26,12 @@
 
 import io.questdb.client.cutlass.qwp.client.GlobalSymbolDictionary;
 import io.questdb.client.cutlass.qwp.client.sf.cursor.PersistedSymbolDict;
+import io.questdb.client.cutlass.qwp.client.sf.cursor.UnreplayableSlotException;
 import io.questdb.client.std.FilesFacade;
+import io.questdb.client.std.MemoryTag;
+import io.questdb.client.std.Crc32c;
 import io.questdb.client.std.ObjList;
+import io.questdb.client.std.Unsafe;
 import io.questdb.client.test.tools.DelegatingFilesFacade;
 import io.questdb.client.test.tools.TestUtils;
 import org.junit.Assert;
@@ -500,6 +504,70 @@ public void testMappedAppendSpansMultipleWindowsAndRecoversDenseIds() throws Exc
         });
     }
 
+    /**
+     * A loaded region that cannot be decoded must surface as an UnreplayableSlotException,
+     * not a raw IllegalStateException.
+     * <p>
+     * Sender.build() routes on the TYPE -- only UnreplayableSlotException reaches its
+     * quarantine handler. A raw IllegalStateException escapes build() instead, and with a
+     * stable senderId and a retained slot every restart re-recovers and rethrows: the
+     * application can never construct a Sender, so it cannot even BUFFER new rows.
+     * <p>
+     * The fixture also documents a real gap: scanRecoveredChunks validates the chunk CRC
+     * and rejects entryCount &lt;= 0, but never checks that entryBytes actually holds
+     * entryCount well-formed entries. A chunk claiming 3 entries inside a 2-byte region is
+     * therefore ACCEPTED here, and only the decoder notices. Validating the triple during
+     * the scan would end the trusted prefix at that chunk instead -- the same treatment a
+     * CRC failure gets.
+     */
+    @Test
+    public void testUndecodableLoadedRegionIsQuarantinableNotAPermanentBrick() throws Exception {
+        assertMemoryLeak(() -> {
+            Path dir = newFolder("qwp-symdict");
+            Path f = dir.resolve(".symbol-dict");
+
+            // [entryCount=3][entryBytes=2][len=1]['a'] -- 3 entries claimed, 1 supplied.
+            byte[] body = new byte[]{0x03, 0x02, 0x01, (byte) 'a'};
+            int crc;
+            long scratch = Unsafe.malloc(body.length, MemoryTag.NATIVE_DEFAULT);
+            try {
+                for (int i = 0; i < body.length; i++) {
+                    Unsafe.getUnsafe().putByte(scratch + i, body[i]);
+                }
+                crc = Crc32c.update(Crc32c.INIT, scratch, body.length);
+            } finally {
+                Unsafe.free(scratch, body.length, MemoryTag.NATIVE_DEFAULT);
+            }
+
+            byte[] file = new byte[8 + body.length + 4];
+            file[0] = 'S';
+            file[1] = 'Y';
+            file[2] = 'D';
+            file[3] = '1';
+            file[4] = 3; // VERSION
+            System.arraycopy(body, 0, file, 8, body.length);
+            int crcAt = 8 + body.length;
+            file[crcAt] = (byte) crc;
+            file[crcAt + 1] = (byte) (crc >>> 8);
+            file[crcAt + 2] = (byte) (crc >>> 16);
+            file[crcAt + 3] = (byte) (crc >>> 24);
+            Files.write(f, file);
+
+            try (PersistedSymbolDict dict = PersistedSymbolDict.open(dir.toString())) {
+                Assert.assertNotNull("the CRC is valid, so the chunk is accepted by the scan", dict);
+                Assert.assertEquals("the scan trusts the declared entryCount", 3, dict.size());
+                try {
+                    dict.addLoadedSymbolsTo(new GlobalSymbolDictionary());
+                    Assert.fail("decoding 3 entries out of a 2-byte region must fail");
+                } catch (UnreplayableSlotException expected) {
+                    Assert.assertTrue("expected the truncated-entry message, got: "
+                                    + expected.getMessage(),
+                            expected.getMessage().contains("truncated loaded symbol dictionary entry"));
+                }
+            }
+        });
+    }
+
     private static int varintSize(int v) {
         int n = 1;
         while ((v >>>= 7) != 0) {