Close the untested gaps around the symbol-dict guards

Four behaviours the delta symbol-dictionary work relies on had no test
that could fail if they regressed. Each mutation below leaves the whole
suite green today.

Pin the split path's write-ahead ordering. Moving the persist after the
publish in flushPendingRowsSplit left every test passing, because the
only split-path test checked the dictionary after a SUCCESSFUL flush --
where the symbols land either way. Only a publish that FAILS while the
symbols are new can tell the orderings apart, so the new test caps the
server batch size to force the split and shrinks the segment so the
split frame cannot be published. With the ordering broken, a process
crash in that window would leave a recorded frame whose deltaStart
exceeds the recovered dictionary, and recovery would quarantine a slot
that should have drained cleanly.

Pin the persist-failure translation. A short write to .symbol-dict --
a full disk, an exhausted quota -- throws a low-level
IllegalStateException that persistNewSymbolsBeforePublish wraps in a
LineSenderException. Deleting the wrap left the suite green, yet the
raw exception would sail past every caller's catch around flush().
Nothing could reach that path: PersistedSymbolDict has facade-aware
overloads but CursorSendEngine called only the FilesFacade.INSTANCE
ones, so the engine now takes a FilesFacade and the test drives a real
short write through the real producer path.

Pin the sealed-segment walk. Deleting the sealed half of
SegmentRing.maxSymbolDeltaEnd left the suite green because every test
fits its slot in one active segment. It is not a benign wrong answer:
recoveredMaxSymbolId collapses to -1, and the seed guard fires on
`>= pd.size()`, so a value that is too low never fires at all -- a torn
dictionary is trusted and the producer silently reuses ids the
surviving frames already define. The walk earns its keep exactly when
the active segment holds nothing but an aborted deferred tail, which is
the crash the guard was built for, so the test reproduces that shape.

Hoist DelegatingFilesFacade into test/tools so a fault seam can be
shared rather than stamped into each test that needs one.

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/CursorSendEngine.java

@@ -27,8 +27,10 @@
 import io.questdb.client.cutlass.qwp.protocol.QwpConstants;
 import io.questdb.client.std.Compat;
 import io.questdb.client.std.Files;
+import io.questdb.client.std.FilesFacade;
 import io.questdb.client.std.ObjList;
 import io.questdb.client.std.QuietCloseable;
+import org.jetbrains.annotations.TestOnly;
 
 import java.util.concurrent.locks.LockSupport;
 
@@ -154,9 +156,29 @@ public CursorSendEngine(String sfDir, long segmentSizeBytes) {
      */
     public CursorSendEngine(String sfDir, long segmentSizeBytes,
                             long maxTotalBytes, long appendDeadlineNanos) {
+        this(sfDir, segmentSizeBytes, maxTotalBytes, appendDeadlineNanos, FilesFacade.INSTANCE);
+    }
+
+    /**
+     * As {@link #CursorSendEngine(String, long, long, long)}, but with an explicit
+     * {@link FilesFacade} for the persisted symbol dictionary.
+     * <p>
+     * The seam exists so a test can drive a dictionary I/O fault -- a short write from
+     * a full disk or an exhausted quota -- through the REAL producer path
+     * ({@code flush()} -> the write-ahead persist), and assert it surfaces as a
+     * {@code LineSenderException} like every other flush-path failure rather than as a
+     * raw {@code IllegalStateException} that would sail past every caller's
+     * {@code catch (LineSenderException)}. Nothing else could reach that translation:
+     * {@code PersistedSymbolDict} has facade-aware overloads, but the engine used to
+     * call only the {@code FilesFacade.INSTANCE} ones, so no fault could be injected
+     * from outside.
+     */
+    @TestOnly
+    public CursorSendEngine(String sfDir, long segmentSizeBytes,
+                            long maxTotalBytes, long appendDeadlineNanos, FilesFacade dictFf) {
         this(sfDir, segmentSizeBytes,
                 new SegmentManager(segmentSizeBytes, SegmentManager.DEFAULT_POLL_NANOS, maxTotalBytes),
-                true, appendDeadlineNanos);
+                true, appendDeadlineNanos, dictFf);
     }
 
     /**
@@ -165,11 +187,12 @@ public CursorSendEngine(String sfDir, long segmentSizeBytes,
      * ownership of the manager. Uses the default append deadline.
      */
     public CursorSendEngine(String sfDir, long segmentSizeBytes, SegmentManager manager) {
-        this(sfDir, segmentSizeBytes, manager, false, DEFAULT_APPEND_DEADLINE_NANOS);
+        this(sfDir, segmentSizeBytes, manager, false, DEFAULT_APPEND_DEADLINE_NANOS,
+                FilesFacade.INSTANCE);
     }
 
     private CursorSendEngine(String sfDir, long segmentSizeBytes, SegmentManager manager,
-                             boolean ownsManager, long appendDeadlineNanos) {
+                             boolean ownsManager, long appendDeadlineNanos, FilesFacade dictFf) {
         // sfDir == null  → memory-only mode (non-SF async ingest). Same
         //                  cursor architecture, no disk involvement; segments
         //                  live in malloc'd native memory.
@@ -278,7 +301,7 @@ private CursorSendEngine(String sfDir, long segmentSizeBytes, SegmentManager man
                 // Load the persisted symbol dictionary so delta-encoded frames
                 // in this recovered slot can be re-registered on the fresh
                 // server before replay. Null on open failure -> delta disabled.
-                persistedDictInProgress = PersistedSymbolDict.open(sfDir);
+                persistedDictInProgress = PersistedSymbolDict.open(dictFf, sfDir);
                 long baseSeed = lowestBase - 1;
                 long watermarkFsn = watermarkInProgress != null
                         ? watermarkInProgress.read()
@@ -389,7 +412,7 @@ private CursorSendEngine(String sfDir, long segmentSizeBytes, SegmentManager man
                     // Windows share lock); if the clean open itself fails,
                     // persistedSymbolDict stays null and the sender falls back to full
                     // self-sufficient frames, which is also safe.
-                    persistedDictInProgress = PersistedSymbolDict.openClean(sfDir);
+                    persistedDictInProgress = PersistedSymbolDict.openClean(dictFf, sfDir);
                 }
                 MmapSegment initial;
                 String initialPath = null;

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

@@ -26,6 +26,9 @@
 
 import io.questdb.client.Sender;
 import io.questdb.client.cutlass.line.LineSenderException;
+import io.questdb.client.cutlass.qwp.client.QwpWebSocketSender;
+import io.questdb.client.cutlass.qwp.client.sf.cursor.CursorSendEngine;
+import io.questdb.client.test.tools.DelegatingFilesFacade;
 import io.questdb.client.cutlass.qwp.client.sf.cursor.PersistedSymbolDict;
 import io.questdb.client.test.cutlass.qwp.websocket.TestWebSocketServer;
 import io.questdb.client.test.tools.TestUtils;
@@ -580,6 +583,56 @@ public void testCommitMessageDoesNotShipUnpersistedLeakedSymbol() throws Excepti
         });
     }
 
+    @Test
+    public void testPersistFailureSurfacesAsLineSenderException() throws Exception {
+        // A dictionary write that fails mid-flush -- a full disk, an exhausted quota --
+        // must reach the caller as a LineSenderException, like every other flush-path
+        // failure. PersistedSymbolDict throws a low-level IllegalStateException on a short
+        // write; persistNewSymbolsBeforePublish wraps it. Without the wrap the raw
+        // IllegalStateException sails straight past every user's
+        // `catch (LineSenderException)` around flush() and takes the application down.
+        //
+        // Nothing could reach that translation before: PersistedSymbolDict has
+        // facade-aware overloads, but CursorSendEngine called only the
+        // FilesFacade.INSTANCE ones, so no test could inject a dictionary I/O fault
+        // through the real producer path. The engine now takes a FilesFacade for exactly
+        // this, and the sender is built on that engine directly -- the same
+        // QwpWebSocketSender.connect(..., CursorSendEngine) entry point Sender.build uses.
+        assertMemoryLeak(() -> {
+            try (TestWebSocketServer silent = new TestWebSocketServer(new SilentHandler())) {
+                int port = silent.getPort();
+                silent.start();
+                Assert.assertTrue(silent.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));
+                ShortDictWriteFacade ff = new ShortDictWriteFacade();
+                // The engine owns the dictionary; the fault facade reaches only it, so the
+                // segment files still write normally and the ONLY failure is the persist.
+                CursorSendEngine engine = new CursorSendEngine(
+                        slot, 4L * 1024 * 1024, CursorSendEngine.DEFAULT_APPEND_DEADLINE_NANOS,
+                        CursorSendEngine.DEFAULT_APPEND_DEADLINE_NANOS, ff);
+                try (Sender sender = QwpWebSocketSender.connect(
+                        "localhost", port, null, 0, 0, 0L, null, false, engine)) {
+                    ff.armed = true; // the next dictionary append short-writes
+                    sender.table("m").symbol("s", "boom").longColumn("v", 1L).atNow();
+                    try {
+                        sender.flush();
+                        Assert.fail("a short write to the symbol dictionary must fail the flush");
+                    } catch (LineSenderException expected) {
+                        Assert.assertTrue("the persist failure must be reported as a sender error, "
+                                        + "not leak a raw IllegalStateException: " + expected.getMessage(),
+                                expected.getMessage().contains("failed to persist symbol dictionary before publish"));
+                    }
+                } catch (LineSenderException ignored) {
+                    // close() re-flushes the still-buffered row; the facade has disarmed, so
+                    // this normally succeeds. Either way it is not what we assert.
+                }
+            }
+        });
+    }
+
     @Test
     public void testFailedPublishDoesNotDuplicatePersistedSymbols() throws Exception {
         // Regression: persistNewSymbolsBeforePublish is a write-ahead -- it runs
@@ -1256,4 +1309,22 @@ private static byte[] buildAck(long seq) {
             return buf;
         }
     }
+    /**
+     * Short-writes the FIRST dictionary append after it is armed, modelling a disk that
+     * fills mid-flush. Offset 0 is left alone so the file header still writes -- only the
+     * entry append fails, which is the path under test.
+     */
+    private static final class ShortDictWriteFacade extends DelegatingFilesFacade {
+        boolean armed;
+
+        @Override
+        public long write(int fd, long addr, long len, long offset) {
+            if (armed && offset > 0 && len > 1) {
+                armed = false;
+                return INSTANCE.write(fd, addr, len - 1, offset);
+            }
+            return INSTANCE.write(fd, addr, len, offset);
+        }
+    }
+
 }

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

@@ -26,6 +26,8 @@
 
 import io.questdb.client.Sender;
 import io.questdb.client.cutlass.line.LineSenderException;
+import io.questdb.client.std.ObjList;
+import io.questdb.client.cutlass.qwp.client.sf.cursor.PersistedSymbolDict;
 import io.questdb.client.test.cutlass.qwp.websocket.TestWebSocketServer;
 import io.questdb.client.test.tools.TestUtils;
 import org.junit.Assert;
@@ -332,6 +334,89 @@ public void testOversizedTableSplitStrandsNothing() throws Exception {
         });
     }
 
+    @Test
+    public void testFileModeSplitPersistsDictBeforeAFAILEDPublish() throws Exception {
+        // The SPLIT path's write-ahead ordering: persistNewSymbolsBeforePublish must run
+        // BEFORE sealAndSwapBuffer publishes the frame to the ring, exactly as it does on
+        // the non-split path.
+        //
+        // Its sibling below (testFileModeSplitPersistsDictBeforePublish) checks the dict
+        // after a SUCCESSFUL flush, which proves nothing about ORDER -- the symbols land
+        // in the file either way. Only a publish that FAILS while the symbols are new can
+        // tell the two apart: with the write-ahead intact the symbols are already durable
+        // when the publish throws; with the persist moved after the publish, the throw
+        // beats it and the dictionary is still empty. (The non-split path is pinned this
+        // way by DeltaDictRecoveryTest.testFailedPublishDoesNotDuplicatePersistedSymbols;
+        // the split path had no equivalent.)
+        //
+        // Why it matters: store-and-forward is process-crash durable. If a split frame
+        // reaches the ring before its symbols reach .symbol-dict, a JVM crash in that
+        // window leaves a recorded frame whose deltaStart exceeds the recovered
+        // dictionary -- so recovery declares the slot unreplayable and quarantines a slot
+        // that should have drained cleanly.
+        //
+        // Setup: the server cap (150) splits the two-table batch, and each split frame
+        // still FITS that cap, so the split pre-flight passes and the publish is actually
+        // attempted. sf_max_bytes=64 then makes every frame larger than the segment, so
+        // appendBlocking fails with PAYLOAD_TOO_LARGE -- deterministically, no
+        // backpressure timing needed.
+        Path sfDir = Files.createTempDirectory("qwp-sf-split-persist-fail");
+        try {
+            assertMemoryLeak(() -> {
+                CapturingHandler handler = new CapturingHandler();
+                try (TestWebSocketServer server = new TestWebSocketServer(handler)) {
+                    server.setAdvertisedMaxBatchSize(150); // forces the two-table batch to split
+                    int port = server.getPort();
+                    server.start();
+                    Assert.assertTrue(server.awaitStart(5, TimeUnit.SECONDS));
+
+                    String config = "ws::addr=localhost:" + port + ";sf_dir=" + sfDir
+                            + ";sf_max_bytes=64;";
+                    String pad = TestUtils.repeat("x", 60);
+                    Sender sender = Sender.fromConfig(config);
+                    try {
+                        sender.table("t1").symbol("s", "alpha").stringColumn("p", pad).longColumn("v", 1L).atNow();
+                        sender.table("t2").symbol("s", "bravo").stringColumn("p", pad).longColumn("v", 2L).atNow();
+                        try {
+                            sender.flush();
+                            Assert.fail("a split frame larger than the segment must fail to publish");
+                        } catch (LineSenderException expected) {
+                            // PAYLOAD_TOO_LARGE -- the publish, not the pre-flight.
+                        }
+                        Assert.assertEquals("the publish must fail, so no frame reaches the server",
+                                0, handler.batches.size());
+
+                        // The write-ahead already ran: both of the batch's new symbols are
+                        // durable even though the frame that references them never
+                        // published. Move the persist after sealAndSwapBuffer and this is 0.
+                        PersistedSymbolDict pd = PersistedSymbolDict.open(
+                                sfDir.resolve("default").toString());
+                        Assert.assertNotNull(pd);
+                        try {
+                            Assert.assertEquals("the split path must persist its new symbols BEFORE "
+                                    + "publishing the frame that references them", 2, pd.size());
+                            ObjList<String> symbols = pd.readLoadedSymbols();
+                            Assert.assertEquals("alpha", symbols.getQuick(0));
+                            Assert.assertEquals("bravo", symbols.getQuick(1));
+                        } finally {
+                            pd.close();
+                        }
+                    } finally {
+                        try {
+                            sender.close();
+                        } catch (LineSenderException ignored) {
+                            // close() re-flushes the still-buffered oversized rows and fails
+                            // again; expected, and not what this test asserts. close() still
+                            // runs its resource cleanup, so no native memory leaks.
+                        }
+                    }
+                }
+            });
+        } finally {
+            rmDir(sfDir);
+        }
+    }
+
     @Test
     public void testFileModeSplitPersistsDictBeforePublish() throws Exception {
         // File-mode store-and-forward + a SPLIT flush: a two-table batch whose combined

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

@@ -31,6 +31,7 @@
 import io.questdb.client.cutlass.qwp.client.sf.cursor.SegmentManager;
 import io.questdb.client.cutlass.qwp.client.sf.cursor.SlotLock;
 import io.questdb.client.std.Files;
+import io.questdb.client.std.FilesFacade;
 import io.questdb.client.std.MemoryTag;
 import io.questdb.client.std.ObjList;
 import io.questdb.client.std.Unsafe;
@@ -608,11 +609,12 @@ private static void assertSlotCanBeReacquired(String sfDir) {
     private static Throwable invokeOwnedPrivateConstructorExpectingFailure(
             String sfDir, long segmentSizeBytes, SegmentManager manager) throws Exception {
         Constructor<CursorSendEngine> ctor = CursorSendEngine.class.getDeclaredConstructor(
-                String.class, long.class, SegmentManager.class, boolean.class, long.class);
+                String.class, long.class, SegmentManager.class, boolean.class, long.class,
+                FilesFacade.class);
         ctor.setAccessible(true);
         try {
             ctor.newInstance(sfDir, segmentSizeBytes, manager, true,
-                    CursorSendEngine.DEFAULT_APPEND_DEADLINE_NANOS);
+                    CursorSendEngine.DEFAULT_APPEND_DEADLINE_NANOS, FilesFacade.INSTANCE);
             fail("expected constructor failure");
             return null;
         } catch (InvocationTargetException e) {