Tidy delta symbol-dict dead code and edges

Minor follow-ups from the review.

Remove the unused CursorSendEngine.isMemoryMode() -- it had no callers;
isDeltaDictEnabled() tests sfDir == null directly.

Sort the NativeBufferWriter import into its alphabetical place among the
qwp.client imports in CursorWebSocketSendLoop.

Harden PersistedSymbolDict.openExisting: hoist entriesAddr / entriesLen
so the catch frees the loaded-entries buffer too. It is unreachable today
(nothing between that malloc and the return throws, and on success the
buffer is transferred to the returned dict), but the error path no longer
leaks if a future edit adds a throwing step.

Return a defensive copy from DeltaDictCatchUpTest's dictFor() instead of
the live inner list: the caller iterates it unlocked while the server
thread may still be appending, matching the sibling dictSnapshot().

Add testTornDictionaryOneIdGapFailsCleanly -- the tightest torn-dictionary
boundary (deltaStart == recoveredSize + 1), the common one-entry-short
host-crash tail. The existing torn test uses a 3-id gap, so it does not
pin the guard's false-negative edge; a "deltaStart > size + 1" mutation
passes it but ships the gapped frame here (the new test fails on 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/CursorSendEngine.java

@@ -629,17 +629,6 @@ public long getTotalBackpressureStalls() {
         return backpressureStallCount.get();
     }
 
-    /**
-     * True when this engine has no store-and-forward directory: the ring lives
-     * only in malloc'd memory, so it cannot be recovered after a crash and no
-     * orphan drainer ever replays it. Only in-process reconnect/failover replays
-     * its frames, which is what makes send-time symbol-dict catch-up (rather than
-     * fully self-sufficient frames) safe.
-     */
-    public boolean isMemoryMode() {
-        return sfDir == null;
-    }
-
     /**
      * Whether the sender may delta-encode symbol dictionaries on this engine.
      * Always true in memory mode (the send loop keeps an in-process catch-up

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

@@ -31,12 +31,12 @@
 import io.questdb.client.cutlass.http.client.WebSocketFrameHandler;
 import io.questdb.client.cutlass.http.client.WebSocketUpgradeException;
 import io.questdb.client.cutlass.line.LineSenderException;
+import io.questdb.client.cutlass.qwp.client.NativeBufferWriter;
 import io.questdb.client.cutlass.qwp.client.QwpAuthFailedException;
 import io.questdb.client.cutlass.qwp.client.QwpDurableAckMismatchException;
 import io.questdb.client.cutlass.qwp.client.QwpIngressRoleRejectedException;
 import io.questdb.client.cutlass.qwp.client.QwpRoleMismatchException;
 import io.questdb.client.cutlass.qwp.client.QwpVersionMismatchException;
-import io.questdb.client.cutlass.qwp.client.NativeBufferWriter;
 import io.questdb.client.cutlass.qwp.client.WebSocketResponse;
 import io.questdb.client.cutlass.qwp.protocol.QwpConstants;
 import io.questdb.client.cutlass.qwp.websocket.WebSocketCloseCode;

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

@@ -353,6 +353,8 @@ private static PersistedSymbolDict openExisting(String filePath, long fileLen) {
             return null;
         }
         long buf = 0L;
+        long entriesAddr = 0L;
+        int entriesLen = 0;
         try {
             int len = (int) fileLen;
             buf = Unsafe.malloc(len, MemoryTag.NATIVE_DEFAULT);
@@ -388,8 +390,7 @@ private static PersistedSymbolDict openExisting(String filePath, long fileLen) {
                 pos = next + (int) entryLen;
                 count++;
             }
-            int entriesLen = pos - HEADER_SIZE;
-            long entriesAddr = 0L;
+            entriesLen = pos - HEADER_SIZE;
             if (entriesLen > 0) {
                 entriesAddr = Unsafe.malloc(entriesLen, MemoryTag.NATIVE_DEFAULT);
                 Unsafe.getUnsafe().copyMemory(buf + HEADER_SIZE, entriesAddr, entriesLen);
@@ -417,6 +418,13 @@ private static PersistedSymbolDict openExisting(String filePath, long fileLen) {
             if (buf != 0L) {
                 Unsafe.free(buf, (int) fileLen, MemoryTag.NATIVE_DEFAULT);
             }
+            // entriesAddr is transferred to the returned dict on the success path,
+            // and the catch only runs before that return, so freeing it here cannot
+            // double-free. Unreachable today (nothing between its malloc and the
+            // return throws), but keeps the error path leak-free against a future edit.
+            if (entriesAddr != 0L) {
+                Unsafe.free(entriesAddr, entriesLen, MemoryTag.NATIVE_DEFAULT);
+            }
             Files.close(fd);
             LOG.warn("symbol dict {} recovery failed ({}); recreating", filePath, String.valueOf(t));
             return null;

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

@@ -272,7 +272,9 @@ private static class CatchUpHandler implements TestWebSocketServer.WebSocketServ
 
         synchronized List<String> dictFor(int connNumber) {
             return connNumber <= dictsByConn.size()
-                    ? dictsByConn.get(connNumber - 1)
+                    // Copy under the lock: the caller iterates it unlocked while the
+                    // server thread may still be appending to the live inner list.
+                    ? new ArrayList<>(dictsByConn.get(connNumber - 1))
                     : new ArrayList<>();
         }
 
@@ -406,7 +408,9 @@ private static class SplitCatchUpHandler implements TestWebSocketServer.WebSocke
 
         synchronized List<String> dictFor(int connNumber) {
             return connNumber <= dictsByConn.size()
-                    ? dictsByConn.get(connNumber - 1)
+                    // Copy under the lock: the caller iterates it unlocked while the
+                    // server thread may still be appending to the live inner list.
+                    ? new ArrayList<>(dictsByConn.get(connNumber - 1))
                     : new ArrayList<>();
         }
 

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

@@ -227,6 +227,81 @@ public void testTornDictionaryFailsCleanlyInsteadOfCorrupting() throws Exception
         });
     }
 
+    @Test
+    public void testTornDictionaryOneIdGapFailsCleanly() throws Exception {
+        // Boundary variant of testTornDictionaryFailsCleanlyInsteadOfCorrupting: the
+        // recovered dictionary is short by EXACTLY ONE id -- the tightest gap the
+        // guard must still reject (deltaStart == recoveredSize + 1). A one-entry-short
+        // tail is the common host-crash outcome, so this pins the guard's
+        // false-negative edge: a "deltaStart > size + 1" mutation would let this
+        // single-id gap through and null-pad the missing symbol on the server.
+        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();
+                    }
+                }
+            }
+
+            // Lose the whole dictionary (truncate to the 8-byte header, size 0) but
+            // stamp the watermark at FSN 0, so recovery replays from FSN 1 -- a frame
+            // with deltaStart=1. The dictionary covers no ids, so id 0 is the single
+            // missing entry: deltaStart(1) == recoveredSize(0) + 1. Because size is 0
+            // no catch-up is sent, so any counted frame would be the gapped data frame.
+            java.nio.file.Path slot = Paths.get(sfDir, "default");
+            java.nio.file.Path dict = slot.resolve(".symbol-dict");
+            byte[] header = Arrays.copyOf(java.nio.file.Files.readAllBytes(dict), 8);
+            java.nio.file.Files.write(dict, header);
+            writeAckWatermark(slot.resolve(".ack-watermark"), 0);
+
+            // Phase 2: recover against a fresh counting server. The guard must fire on
+            // the very first replay frame (deltaStart 1 > recovered size 0) and fail
+            // terminally rather than send a frame that null-pads id 0.
+            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("a one-id gap must still block replay to a fresh server",
+                        0, handler.frames.get());
+                Assert.assertNotNull("a one-id-short dictionary must surface a terminal error", terminal);
+                Assert.assertTrue(terminal.getMessage(),
+                        terminal.getMessage().contains("delta start 1 exceeds recovered dictionary size 0"));
+            }
+        });
+    }
+
     @Test
     public void testRecoveredSenderContinuesIngestingNewSymbols() throws Exception {
         // M2 regression: seedGlobalDictionaryFromPersisted resumes the producer's