Plug ensureConnected leak; consolidate varint helpers

Review cleanups:

- ensureConnected's start()-failure catch now closes cursorSendLoop
  before nulling client. Previously it closed only the client; because
  the connected flag is set past the catch, a caller that retries --
  re-entering ensureConnected and reassigning cursorSendLoop -- orphaned
  a recovered slot's ctor-seeded native mirror. That is a reachable leak
  on the recovered-slot + start()/dispatcher-OOM + retry path. close() is
  idempotent and frees the mirror via its loopNeverRan path, and also
  closes the shared client, so the following client.close() is a no-op.

- Consolidated the triplicated raw-address varint writer into
  NativeBufferWriter.writeVarint (writeVarintDirect now delegates to it);
  PersistedSymbolDict and the catch-up frame builder in
  CursorWebSocketSendLoop use it, and PersistedSymbolDict.varintSize now
  reuses NativeBufferWriter.varintSize.

- Alphabetized CursorSendEngine (getPersistedSymbolDict before
  getTotalBackpressureStalls before isDeltaDictEnabled) and moved
  PersistedSymbolDict.decodeVarint to the front of its private-static
  cluster.

- readVarintAt gained the shift > 35 defensive bound that decodeVarint
  already has (unreachable today, all inputs are freshly-encoded or
  validated varints, but consistent).

- CursorWebSocketSendLoopCatchUpAlignmentTest now uses the existing
  TestUtils.createTmpDir/removeTmpDir instead of reinventing them.

Not changed: the reported dead CursorSendEngine.isMemoryMode() and a
PersistedSymbolDict MAX_ENTRY_LEN constant do not exist at HEAD (the
former removed earlier on this branch, the latter never present), and the
NativeBufferWriter import is already correctly ordered. The
buildAck/waitFor/rmDir/readVarint test helpers are duplicated across ~19
pre-existing test files, so consolidating them belongs in a dedicated
repo-wide test-hygiene change rather than this feature PR.

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

Author: glasstiger

core/src/main/java/io/questdb/client/cutlass/qwp/client/NativeBufferWriter.java

@@ -76,6 +76,21 @@ public static int varintSize(long value) {
         return (64 - Long.numberOfLeadingZeros(value) + 6) / 7;
     }
 
+    /**
+     * Writes {@code value} as an unsigned LEB128 varint directly at native address
+     * {@code addr} and returns the address just past the last byte. The canonical
+     * raw-address varint writer shared by the SF cursor's persisted dictionary and
+     * catch-up frame builder.
+     */
+    public static long writeVarint(long addr, long value) {
+        while (value > 0x7F) {
+            Unsafe.getUnsafe().putByte(addr++, (byte) ((value & 0x7F) | 0x80));
+            value >>>= 7;
+        }
+        Unsafe.getUnsafe().putByte(addr++, (byte) value);
+        return addr;
+    }
+
     @Override
     public void close() {
         if (bufferPtr != 0) {
@@ -336,11 +351,7 @@ public void skip(int bytes) {
     }
 
     private static void writeVarintDirect(long addr, long value) {
-        while (value > 0x7F) {
-            Unsafe.getUnsafe().putByte(addr++, (byte) ((value & 0x7F) | 0x80));
-            value >>>= 7;
-        }
-        Unsafe.getUnsafe().putByte(addr, (byte) value);
+        writeVarint(addr, value);
     }
 
     private void encodeUtf8(CharSequence value, int utf8Len) {

core/src/main/java/io/questdb/client/cutlass/qwp/client/QwpWebSocketSender.java

@@ -3362,6 +3362,17 @@ private void ensureConnected() {
             cursorSendLoop.setConnectionDispatcher(connectionDispatcher);
             cursorSendLoop.start();
         } catch (Throwable t) {
+            // start() (or dispatcher construction) failed after cursorSendLoop was
+            // assigned. Close it so a caller that retries -- re-entering
+            // ensureConnected and reassigning cursorSendLoop above -- cannot orphan
+            // a recovered slot's ctor-seeded native mirror (freed only by close()
+            // or the I/O loop, neither of which has run). close() is idempotent and
+            // frees the mirror via its loopNeverRan path; it also closes the shared
+            // client, so the client.close() below is a safe idempotent no-op.
+            if (cursorSendLoop != null) {
+                cursorSendLoop.close();
+                cursorSendLoop = null;
+            }
             if (client != null) {
                 client.close();
                 client = null;

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

@@ -619,6 +619,15 @@ public MmapSegment firstSealed() {
         return ring.firstSealed();
     }
 
+    /**
+     * The engine's persisted symbol dictionary, or {@code null} in memory mode
+     * (and in disk mode if it failed to open). The producer appends new symbols
+     * to it; recovery / orphan-drain read its loaded entries to seed catch-up.
+     */
+    public PersistedSymbolDict getPersistedSymbolDict() {
+        return persistedSymbolDict;
+    }
+
     /**
      * Number of times {@link #appendBlocking} hit
      * {@link SegmentRing#BACKPRESSURE_NO_SPARE} on its first attempt and
@@ -641,15 +650,6 @@ public boolean isDeltaDictEnabled() {
         return sfDir == null || persistedSymbolDict != null;
     }
 
-    /**
-     * The engine's persisted symbol dictionary, or {@code null} in memory mode
-     * (and in disk mode if it failed to open). The producer appends new symbols
-     * to it; recovery / orphan-drain read its loaded entries to seed catch-up.
-     */
-    public PersistedSymbolDict getPersistedSymbolDict() {
-        return persistedSymbolDict;
-    }
-
     /**
      * Pass-through to {@link SegmentRing#nextSealedAfter(MmapSegment)}.
      */

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

@@ -2161,6 +2161,14 @@ private long readVarintAt(long p, long limit) {
                 break;
             }
             shift += 7;
+            if (shift > 35) {
+                // Defensive bound, matching PersistedSymbolDict.decodeVarint: a
+                // canonical entry-length / delta varint is <= 5 bytes. Every caller
+                // reads freshly-encoded, CRC- or openExisting-validated bytes, so
+                // this is unreachable, but it stops a corrupt continuation run from
+                // over-shifting into a garbage length.
+                break;
+            }
         }
         varintEnd = cur;
         return value;
@@ -2310,8 +2318,8 @@ private void sendCatchUpChunk(int deltaStart, int deltaCount, long symbolsAddr,
                             | QwpConstants.FLAG_DEFER_COMMIT));
             Unsafe.getUnsafe().putShort(frame + 6, (short) 0); // tableCount
             Unsafe.getUnsafe().putInt(frame + 8, payloadLen);
-            long q = writeVarintAt(frame + QwpConstants.HEADER_SIZE, deltaStart);
-            q = writeVarintAt(q, deltaCount);
+            long q = NativeBufferWriter.writeVarint(frame + QwpConstants.HEADER_SIZE, deltaStart);
+            q = NativeBufferWriter.writeVarint(q, deltaCount);
             Unsafe.getUnsafe().copyMemory(symbolsAddr, q, symbolsLen);
             client.sendBinary(frame, frameLen);
         } catch (Throwable t) {
@@ -2331,15 +2339,6 @@ private void sendCatchUpChunk(int deltaStart, int deltaCount, long symbolsAddr,
         totalFramesSent.incrementAndGet();
     }
 
-    private static long writeVarintAt(long addr, long value) {
-        while (value > 0x7F) {
-            Unsafe.getUnsafe().putByte(addr++, (byte) ((value & 0x7F) | 0x80));
-            value >>>= 7;
-        }
-        Unsafe.getUnsafe().putByte(addr++, (byte) value);
-        return addr;
-    }
-
     private boolean tryReceiveAcks() {
         boolean any = false;
         try {

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

@@ -25,6 +25,7 @@
 package io.questdb.client.cutlass.qwp.client.sf.cursor;
 
 import io.questdb.client.cutlass.qwp.client.GlobalSymbolDictionary;
+import io.questdb.client.cutlass.qwp.client.NativeBufferWriter;
 import io.questdb.client.std.Files;
 import io.questdb.client.std.MemoryTag;
 import io.questdb.client.std.ObjList;
@@ -218,10 +219,10 @@ public synchronized void appendSymbol(CharSequence symbol) {
             return;
         }
         int utf8Len = Utf8s.utf8Bytes(symbol);
-        int varLen = varintSize(utf8Len);
+        int varLen = NativeBufferWriter.varintSize(utf8Len);
         int recLen = varLen + utf8Len;
         ensureScratch(recLen);
-        long p = writeVarint(scratchAddr, utf8Len);
+        long p = NativeBufferWriter.writeVarint(scratchAddr, utf8Len);
         if (utf8Len > 0) {
             Utf8s.strCpyUtf8(symbol, p, utf8Len);
         }
@@ -257,9 +258,9 @@ public synchronized void appendSymbols(GlobalSymbolDictionary dict, int from, in
         for (int id = from; id <= to; id++) {
             CharSequence symbol = dict.getSymbol(id);
             int utf8Len = Utf8s.utf8Bytes(symbol);
-            int recLen = varintSize(utf8Len) + utf8Len;
+            int recLen = NativeBufferWriter.varintSize(utf8Len) + utf8Len;
             ensureScratch(len + recLen);
-            long p = writeVarint(scratchAddr + len, utf8Len);
+            long p = NativeBufferWriter.writeVarint(scratchAddr + len, utf8Len);
             if (utf8Len > 0) {
                 Utf8s.strCpyUtf8(symbol, p, utf8Len);
             }
@@ -351,6 +352,30 @@ public int size() {
         return size;
     }
 
+    /**
+     * Decodes an unsigned LEB128 varint from {@code buf[pos..limit)}. Returns
+     * {@code [value, newPos]} or {@code null} if the varint is truncated
+     * (torn tail).
+     */
+    private static long[] decodeVarint(long buf, int pos, int limit) {
+        long value = 0;
+        int shift = 0;
+        int cur = pos;
+        while (cur < limit) {
+            byte b = Unsafe.getUnsafe().getByte(buf + cur);
+            cur++;
+            value |= (long) (b & 0x7F) << shift;
+            if ((b & 0x80) == 0) {
+                return new long[]{value, cur};
+            }
+            shift += 7;
+            if (shift > 35) {
+                return null; // implausible for an entry length; treat as torn
+            }
+        }
+        return null;
+    }
+
     private static PersistedSymbolDict openExisting(String filePath, long fileLen) {
         int fd = Files.openRW(filePath);
         if (fd < 0) {
@@ -473,48 +498,6 @@ private static PersistedSymbolDict openFresh(String filePath) {
         return new PersistedSymbolDict(fd, HEADER_SIZE, 0, 0L, 0);
     }
 
-    /**
-     * Decodes an unsigned LEB128 varint from {@code buf[pos..limit)}. Returns
-     * {@code [value, newPos]} or {@code null} if the varint is truncated
-     * (torn tail).
-     */
-    private static long[] decodeVarint(long buf, int pos, int limit) {
-        long value = 0;
-        int shift = 0;
-        int cur = pos;
-        while (cur < limit) {
-            byte b = Unsafe.getUnsafe().getByte(buf + cur);
-            cur++;
-            value |= (long) (b & 0x7F) << shift;
-            if ((b & 0x80) == 0) {
-                return new long[]{value, cur};
-            }
-            shift += 7;
-            if (shift > 35) {
-                return null; // implausible for an entry length; treat as torn
-            }
-        }
-        return null;
-    }
-
-    private static int varintSize(long value) {
-        int n = 1;
-        while (value > 0x7F) {
-            value >>>= 7;
-            n++;
-        }
-        return n;
-    }
-
-    private static long writeVarint(long addr, long value) {
-        while (value > 0x7F) {
-            Unsafe.getUnsafe().putByte(addr++, (byte) ((value & 0x7F) | 0x80));
-            value >>>= 7;
-        }
-        Unsafe.getUnsafe().putByte(addr++, (byte) value);
-        return addr;
-    }
-
     private void ensureScratch(int required) {
         if (scratchCap >= required) {
             return;

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

@@ -30,7 +30,6 @@
 import io.questdb.client.cutlass.qwp.client.sf.cursor.CursorSendEngine;
 import io.questdb.client.cutlass.qwp.client.sf.cursor.CursorWebSocketSendLoop;
 import io.questdb.client.network.PlainSocketFactory;
-import io.questdb.client.std.Files;
 import io.questdb.client.std.MemoryTag;
 import io.questdb.client.std.Unsafe;
 import io.questdb.client.test.tools.TestUtils;
@@ -42,7 +41,6 @@
 import java.lang.reflect.InvocationTargetException;
 import java.lang.reflect.Method;
 import java.nio.charset.StandardCharsets;
-import java.nio.file.Paths;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
@@ -78,30 +76,12 @@ public class CursorWebSocketSendLoopCatchUpAlignmentTest {
 
     @Before
     public void setUp() {
-        tmpDir = Paths.get(System.getProperty("java.io.tmpdir"),
-                "qdb-cursor-catchup-" + System.nanoTime()).toString();
-        assertEquals(0, Files.mkdir(tmpDir, Files.DIR_MODE_DEFAULT));
+        tmpDir = TestUtils.createTmpDir("qdb-cursor-catchup-");
     }
 
     @After
     public void tearDown() {
-        if (tmpDir == null) return;
-        long find = Files.findFirst(tmpDir);
-        if (find > 0) {
-            try {
-                int rc = 1;
-                while (rc > 0) {
-                    String name = Files.utf8ToString(Files.findName(find));
-                    if (name != null && !".".equals(name) && !"..".equals(name)) {
-                        Files.remove(tmpDir + "/" + name);
-                    }
-                    rc = Files.findNext(find);
-                }
-            } finally {
-                Files.findClose(find);
-            }
-        }
-        Files.remove(tmpDir);
+        TestUtils.removeTmpDir(tmpDir);
     }
 
     @Test