Harden persisted symbol-dict edge paths

Fold in low-severity hardening found in review; none changes behavior
on any reachable input.

PersistedSymbolDict:
- appendRawEntries fails loudly (rather than flushing an uninitialised
  scratch tail and mis-advancing size) if a caller ever passes an
  (addr, len, count) triple whose entries do not fill len; the sole
  caller derives both from one beginMessage, so this cannot fire today.
- openExisting sets entriesLen alongside the malloc, so the catch frees
  the right size if the copy walk throws -- the comment already claimed
  this held.
- ensureScratch grows in long so scratchCap*2 cannot overflow negative
  past ~1 GB, matching ensureSentDictCapacity.
- Two varint decode loops gain the shift > 35 bound the other decoders
  already carry.

Close a file-descriptor leak in the rmDir test helpers: Files.walk
returns a Stream backed by an open directory handle, so wrap it in
try-with-resources (PersistedSymbolDictTest calls rmDir 13 times).

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

@@ -230,6 +230,12 @@ public synchronized void appendRawEntries(long addr, int len, int count) {
                     break;
                 }
                 shift += 7;
+                if (shift > 35) {
+                    // A canonical entry-length varint is <= 5 bytes; a longer
+                    // continuation run is corrupt. The downstream entryEnd > srcLimit
+                    // check then rejects it. Matches decodeVarint / readVarintAt.
+                    break;
+                }
             }
             long entryEnd = src + symLen; // src is just past the len varint
             if (entryEnd > srcLimit) {
@@ -242,6 +248,16 @@ public synchronized void appendRawEntries(long addr, int len, int count) {
             dst += wireSpan + CRC_SIZE;
             src = entryEnd;
         }
+        if (src != srcLimit) {
+            // The count entries did not consume exactly len bytes -- a caller passed an
+            // inconsistent (addr, len, count) triple. Writing outLen would flush an
+            // uninitialised scratch tail and mis-advance size, so fail loudly. The sole
+            // caller derives count and len from one beginMessage, so this cannot fire
+            // today.
+            throw new IllegalStateException("raw symbol-dict entries under-filled the buffer to "
+                    + FILE_NAME + " [count=" + count + ", len=" + len
+                    + ", consumed=" + (int) (src - addr) + ']');
+        }
         long written = Files.write(fd, scratchAddr, outLen, appendOffset);
         if (written != outLen) {
             throw new IllegalStateException("short write to " + FILE_NAME
@@ -503,6 +519,9 @@ private static PersistedSymbolDict openExisting(String filePath, long fileLen) {
             // open. A second no-alloc walk over the already-validated region.
             if (wireLen > 0) {
                 entriesAddr = Unsafe.malloc(wireLen, MemoryTag.NATIVE_DEFAULT);
+                // Record the length alongside the malloc so the catch below frees the
+                // right size (not 0) if this copy walk ever throws.
+                entriesLen = wireLen;
                 long dst = entriesAddr;
                 int p = HEADER_SIZE;
                 for (int i = 0; i < count; i++) {
@@ -516,14 +535,16 @@ private static PersistedSymbolDict openExisting(String filePath, long fileLen) {
                             break;
                         }
                         shift += 7;
+                        if (shift > 35) {
+                            break; // corrupt run; these entries were CRC-validated above
+                        }
                     }
                     int wireSpan = (vp - p) + (int) symLen; // [len][utf8], no CRC
                     Unsafe.getUnsafe().copyMemory(buf + p, dst, wireSpan);
                     dst += wireSpan;
                     p += wireSpan + CRC_SIZE; // skip the entry's CRC
                 }
             }
-            entriesLen = wireLen;
             Unsafe.free(buf, len, MemoryTag.NATIVE_DEFAULT);
             buf = 0L;
             // Drop any torn/stale trailing bytes so a LATER, shorter append cannot
@@ -601,8 +622,15 @@ private void ensureScratch(int required) {
         if (scratchCap >= required) {
             return;
         }
-        int newCap = Math.max(required, Math.max(256, scratchCap * 2));
-        scratchAddr = Unsafe.realloc(scratchAddr, scratchCap, newCap, MemoryTag.NATIVE_DEFAULT);
-        scratchCap = newCap;
+        // Double in long: scratchCap * 2 as an int overflows negative past ~1 GB and
+        // would make the realloc size negative. required is bounded by one frame's
+        // entries (the server batch cap), so this never actually caps -- it mirrors the
+        // long-math growth in CursorWebSocketSendLoop.ensureSentDictCapacity.
+        long newCap = Math.max(required, Math.max(256L, (long) scratchCap * 2));
+        if (newCap > Integer.MAX_VALUE - 8) {
+            newCap = Integer.MAX_VALUE - 8;
+        }
+        scratchAddr = Unsafe.realloc(scratchAddr, scratchCap, (int) newCap, MemoryTag.NATIVE_DEFAULT);
+        scratchCap = (int) newCap;
     }
 }

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

@@ -38,6 +38,7 @@
 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;
 
@@ -351,15 +352,18 @@ private static void rmDir(Path dir) {
             if (dir == null || !Files.exists(dir)) {
                 return;
             }
-            Files.walk(dir)
-                    .sorted(Comparator.reverseOrder())
-                    .forEach(p -> {
-                        try {
-                            Files.deleteIfExists(p);
-                        } catch (IOException ignored) {
-                            // best-effort
-                        }
-                    });
+            // 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
         }

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

@@ -38,6 +38,7 @@
 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;
 
@@ -199,15 +200,18 @@ private static void rmDir(Path dir) throws IOException {
         if (dir == null || !Files.exists(dir)) {
             return;
         }
-        Files.walk(dir)
-                .sorted(Comparator.reverseOrder())
-                .forEach(p -> {
-                    try {
-                        Files.deleteIfExists(p);
-                    } catch (IOException ignored) {
-                        // best-effort
-                    }
-                });
+        // 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
+                        }
+                    });
+        }
     }
 
     private static class SilentHandler implements TestWebSocketServer.WebSocketServerHandler {

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

@@ -37,6 +37,7 @@
 import java.nio.file.Path;
 import java.nio.file.StandardOpenOption;
 import java.util.Comparator;
+import java.util.stream.Stream;
 
 import static io.questdb.client.test.tools.TestUtils.assertMemoryLeak;
 
@@ -503,14 +504,17 @@ private static void rmDir(Path dir) {
             if (dir == null || !Files.exists(dir)) {
                 return;
             }
-            Files.walk(dir)
-                    .sorted(Comparator.reverseOrder())
-                    .forEach(p -> {
-                        try {
-                            Files.deleteIfExists(p);
-                        } catch (IOException ignored) {
-                        }
-                    });
+            // 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) {
+                            }
+                        });
+            }
         } catch (IOException ignored) {
         }
     }