Cover defensive dict guards and tidy test helpers

Round of minor cleanups on the delta symbol-dictionary work.

Tests for previously-uncovered defensive guards (all proven to fail
without their guard): PersistedSymbolDict.open recreates empty on a bad
version byte (a valid entry behind a corrupted version must not parse)
and on a > Integer.MAX_VALUE length (a >2GB dictionary can't map into one
int buffer -- driven by a fault facade); and the send loop's
ensureSentDictCapacity latches a terminal instead of overflowing the int
capacity math when the mirror would exceed MAX_SENT_DICT_BYTES.

openExisting nulls buf after freeing it on the bad-magic/bad-version
path, so the catch block cannot double-free if ff.close ever threw.

PersistedSymbolDict.appendSymbol is marked @testonly -- production
persists a frame's whole new-symbol range via appendSymbols /
appendRawEntries; only tests append one symbol at a time.

Documented that symbol-dict catch-up frames sit outside the poison-frame
detector: a deterministically-NACKed catch-up recycles forever (paced),
which is fine since a catch-up only re-registers already-accepted symbols.

Consolidated four reinvented recursive rmDir/rmDirRec test helpers into
one TestUtils.removeTmpDirRec (the existing removeTmpDir is flat and can't
clean the store-and-forward slot subdirectory layout), and removed the
imports that freed up.

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

@@ -2911,6 +2911,13 @@ private void handleServerRejection(long wireSeq) {
                 // 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.
+                //
+                // Catch-up frames therefore sit OUTSIDE the poison detector: a
+                // deterministically-NACKed catch-up recycles forever (paced, so no
+                // busy-loop). That is acceptable -- a catch-up only re-registers
+                // symbols the cluster already accepted, so a persistent NACK of one
+                // is a server bug, not a poison-frame the client can quarantine, and
+                // Invariant B's "retry a transient outage forever" applies.
                 handlePreSendRejection(wireSeq, status, category, policy);
                 return;
             }

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

@@ -33,6 +33,7 @@
 import io.questdb.client.std.QuietCloseable;
 import io.questdb.client.std.Unsafe;
 import io.questdb.client.std.str.Utf8s;
+import org.jetbrains.annotations.TestOnly;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -317,7 +318,12 @@ public synchronized void appendRawEntries(long addr, int len, int count) {
      * (the entry's position). Writes {@code [len varint][utf8][crc32c]}, the CRC
      * covering the {@code [len][utf8]} bytes so a torn/stale entry is detected on
      * recovery.
+     * <p>
+     * Test-only: production persists a frame's whole new-symbol range in one write
+     * via {@link #appendSymbols} / {@link #appendRawEntries}. Tests use this to
+     * build a dictionary one entry at a time.
      */
+    @TestOnly
     public synchronized void appendSymbol(CharSequence symbol) {
         if (closed) {
             return;
@@ -516,6 +522,7 @@ private static PersistedSymbolDict openExisting(FilesFacade ff, String filePath,
                     || Unsafe.getUnsafe().getByte(buf + 4) != VERSION) {
                 LOG.warn("symbol dict {} unreadable, bad magic or unknown version; recreating", filePath);
                 Unsafe.free(buf, len, MemoryTag.NATIVE_DEFAULT);
+                buf = 0L; // null after free so the catch below cannot double-free if ff.close throws
                 ff.close(fd);
                 return null;
             }

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

@@ -27,7 +27,6 @@
 import io.questdb.client.Sender;
 import io.questdb.client.cutlass.line.LineSenderException;
 import io.questdb.client.cutlass.qwp.client.sf.cursor.PersistedSymbolDict;
-import io.questdb.client.std.Files;
 import io.questdb.client.test.cutlass.qwp.websocket.TestWebSocketServer;
 import io.questdb.client.test.tools.TestUtils;
 import org.junit.After;
@@ -76,9 +75,7 @@ public void setUp() {
 
     @After
     public void tearDown() {
-        if (sfDir != null) {
-            rmDirRec(sfDir);
-        }
+        TestUtils.removeTmpDirRec(sfDir);
     }
 
     @Test
@@ -879,31 +876,6 @@ private static int readVarint(byte[] buf, int[] pos) {
         throw new IllegalStateException("varint truncated");
     }
 
-    private static void rmDirRec(String dir) {
-        if (!Files.exists(dir)) {
-            return;
-        }
-        long find = Files.findFirst(dir);
-        if (find > 0) {
-            try {
-                int rc = 1;
-                while (rc > 0) {
-                    String name = Files.utf8ToString(Files.findName(find));
-                    if (name != null && !".".equals(name) && !"..".equals(name)) {
-                        String child = dir + "/" + name;
-                        if (!Files.remove(child)) {
-                            rmDirRec(child);
-                        }
-                    }
-                    rc = Files.findNext(find);
-                }
-            } finally {
-                Files.findClose(find);
-            }
-        }
-        Files.remove(dir);
-    }
-
     /**
      * Reconstructs the per-connection symbol dictionary from delta sections,
      * mirroring the server's {@code setQuick(deltaStart + i)} + null-padding.

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

@@ -27,6 +27,7 @@
 import io.questdb.client.Sender;
 import io.questdb.client.cutlass.line.LineSenderException;
 import io.questdb.client.test.cutlass.qwp.websocket.TestWebSocketServer;
+import io.questdb.client.test.tools.TestUtils;
 import org.junit.Assert;
 import org.junit.Test;
 
@@ -35,10 +36,8 @@
 import java.nio.ByteOrder;
 import java.nio.file.Files;
 import java.nio.file.Path;
-import java.util.Comparator;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicLong;
-import java.util.stream.Stream;
 
 import static io.questdb.client.test.tools.TestUtils.assertMemoryLeak;
 
@@ -348,25 +347,7 @@ private static int readVarint(byte[] buf, int offset) {
     }
 
     private static void rmDir(Path dir) {
-        try {
-            if (dir == null || !Files.exists(dir)) {
-                return;
-            }
-            // try-with-resources: Files.walk returns a Stream backed by an open
-            // directory handle that must be closed, or each rmDir leaks a descriptor.
-            try (Stream<Path> walk = Files.walk(dir)) {
-                walk.sorted(Comparator.reverseOrder())
-                        .forEach(p -> {
-                            try {
-                                Files.deleteIfExists(p);
-                            } catch (IOException ignored) {
-                                // best-effort
-                            }
-                        });
-            }
-        } catch (IOException ignored) {
-            // best-effort
-        }
+        TestUtils.removeTmpDirRec(dir == null ? null : dir.toString());
     }
 
     private static void waitFor(BoolCondition cond, long timeoutMillis) {

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

@@ -321,6 +321,47 @@ public void testCatchUpCapGapRetriesUntilBudgetThenLatches() throws Exception {
         });
     }
 
+    @Test
+    public void testMirrorOverflowFailsLoud() throws Exception {
+        // ensureSentDictCapacity must latch a terminal -- not silently overflow the
+        // int capacity math into a heap-corrupting copyMemory -- when the sent-dict
+        // mirror would exceed MAX_SENT_DICT_BYTES. Unreachable at real cardinality
+        // (~200M+ symbols on one connection), so drive the guard directly with an
+        // oversized required, mirroring testCatchUpChunkFrameSizeOverflowFailsLoud.
+        TestUtils.assertMemoryLeak(() -> {
+            Field maxField = CursorWebSocketSendLoop.class.getDeclaredField("MAX_SENT_DICT_BYTES");
+            maxField.setAccessible(true);
+            long overCeiling = (long) maxField.getInt(null) + 1L;
+            CatchUpCapturingClient client = new CatchUpCapturingClient(0);
+            try (CursorSendEngine engine = newEngine()) {
+                CursorWebSocketSendLoop loop = newLoop(engine, client);
+                try {
+                    Method m = CursorWebSocketSendLoop.class.getDeclaredMethod("ensureSentDictCapacity", long.class);
+                    m.setAccessible(true);
+                    try {
+                        m.invoke(loop, overCeiling);
+                        fail("a mirror capacity past MAX_SENT_DICT_BYTES must fail loud, not overflow");
+                    } catch (InvocationTargetException e) {
+                        assertEquals("overflow must surface as LineSenderException",
+                                "LineSenderException", e.getCause().getClass().getSimpleName());
+                        assertTrue("message must name the mirror ceiling: " + e.getCause().getMessage(),
+                                e.getCause().getMessage().contains("mirror exceeds the maximum size"));
+                    }
+                    // recordFatal (not a bare throw) latched the terminal, so the loop
+                    // winds down instead of reconnecting into the same overflow.
+                    try {
+                        loop.checkError();
+                        fail("mirror overflow must latch a terminal");
+                    } catch (LineSenderException terminal) {
+                        assertTrue(terminal.getMessage().contains("mirror exceeds the maximum size"));
+                    }
+                } finally {
+                    loop.close();
+                }
+            }
+        });
+    }
+
     private static void appendFrames(CursorSendEngine engine, int count) {
         long buf = Unsafe.malloc(16, MemoryTag.NATIVE_DEFAULT);
         try {

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

@@ -29,16 +29,15 @@
 import io.questdb.client.cutlass.qwp.client.sf.cursor.CursorWebSocketSendLoop;
 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;
 import org.junit.Test;
 
 import java.io.IOException;
 import java.lang.reflect.Field;
 import java.nio.file.Files;
 import java.nio.file.Path;
-import java.util.Comparator;
 import java.util.concurrent.TimeUnit;
-import java.util.stream.Stream;
 
 import static io.questdb.client.test.tools.TestUtils.assertMemoryLeak;
 
@@ -197,21 +196,7 @@ private static int readInt(CursorWebSocketSendLoop loop, String name) throws Exc
     }
 
     private static void rmDir(Path dir) throws IOException {
-        if (dir == null || !Files.exists(dir)) {
-            return;
-        }
-        // try-with-resources: Files.walk returns a Stream backed by an open
-        // directory handle that must be closed, or each rmDir leaks a descriptor.
-        try (Stream<Path> walk = Files.walk(dir)) {
-            walk.sorted(Comparator.reverseOrder())
-                    .forEach(p -> {
-                        try {
-                            Files.deleteIfExists(p);
-                        } catch (IOException ignored) {
-                            // best-effort
-                        }
-                    });
-        }
+        TestUtils.removeTmpDirRec(dir == null ? null : dir.toString());
     }
 
     private static class SilentHandler implements TestWebSocketServer.WebSocketServerHandler {