Close test loops, drop dead walkers, fix misleading comments

Review follow-ups: resource hygiene in the poison-frame tests, dead code
left behind by the recovery fold, and comments that contradict the code.

Wrap the twelve CursorWebSocketSendLoops in
CursorWebSocketSendLoopPoisonFrameTest in try-with-resources. They were
never closed, and their enclosing bodies leaked on the assertion-failure
path too, which TestUtils.assertMemoryLeak hides because it calls
skipChecks() when the body throws. Nesting the loop inside the engine's
try-with-resources also fixes close order: the loop borrows the engine's
persisted dictionary prefix, so it must close first. A probe on all
twelve sites confirms none of them actually leaked -- no symbol-dict
mirror or catch-up frame buffer was allocated on these paths, and the
reconnect factory registers every client in the list closeAll() drains.
The value is in removing the trap: releaseSentDictBytes() and
freeCatchUpFrameBuffer() are reachable only through close(), so a future
test in one of these bodies that triggers a dictionary catch-up would
have leaked silently.

Carry the slot-qualified path into PersistedSymbolDict so the mmap-trim
warning in close() names the dictionary it is about. It logged the bare
FILE_NAME constant, which is unattributable in a process holding many
slots -- a foreground sender plus N orphan drainers.

Delete SegmentRing.findLastFsnWithoutPayloadFlag and
MmapSegment.findLastFrameFsnWithoutPayloadFlag. The recovery fold left
both without callers, and each walks the whole backlog. Drop the two
dead stores in RecoveredFrameAnalysis.finish(): only appendRaw() reads
those fields, and it cannot run after finish().

Correct four comments that state the opposite of what the code does:

- hasEverConnected is documented as observability only while being half
  of endpointPolicyFailureIsTerminal().
- PersistedSymbolDict.close() claims that nulling the pointer prevents
  use-after-free, contradicting loadedEntriesAddr()'s accurate javadoc.
  It guards a construction-phase reader; ordering is what makes the
  borrow safe across threads.
- sentDictBytesOwned describes borrowing as an orphan-only state, but
  loops of either policy borrow now.
- connectWithDurableAckRetry refers to @testonly callers that no longer
  exist.

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

@@ -267,7 +267,7 @@ public BackgroundDrainer() {
      */
     public WebSocketClient connectWithDurableAckRetry() {
         // run() already set runnerThread; setting it again here is a no-op
-        // on that path but wires up direct @TestOnly calls so requestStop()
+        // on that path but wires up direct callers so requestStop()
         // can unpark them too.
         runnerThread = Thread.currentThread();
         long backoffMillis = reconnectInitialBackoffMillis;

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

@@ -299,8 +299,10 @@ public final class CursorWebSocketSendLoop implements QuietCloseable {
     private long sentDictBytesAddr;
     private int sentDictBytesCapacity;
     private int sentDictBytesLen;
-    // False only while an orphan-drainer loop borrows the persisted prefix.
-    // Any growth first performs a copy-on-write; only owned buffers are freed.
+    // False while the mirror is unallocated, and while a loop of EITHER policy borrows
+    // the engine's persisted prefix -- both borrow now; neither consumes it. Any growth
+    // copy-on-writes into loop-owned memory and sets this true; only owned buffers are
+    // freed.
     private boolean sentDictBytesOwned;
     private int sentDictCount;
     // True when replay frames can start above dictionary id zero and therefore
@@ -373,8 +375,12 @@ public final class CursorWebSocketSendLoop implements QuietCloseable {
     private long fsnAtZero;
     // Sticky flag: false until the very first time a live client is installed
     // (either via the constructor in SYNC/OFF mode or via swapClient on a
-    // successful connect attempt in any mode). Once true, stays true. This is
-    // connection-state observability only; reconnect policy is role-based.
+    // successful connect attempt in any mode). Once true, stays true.
+    // LOAD-BEARING, not observability: it is half of endpointPolicyFailureIsTerminal(),
+    // which latches an auth / upgrade / durable-ack rejection only for an ORPHAN drainer
+    // or a sender that has never connected. Relocate or lazily initialise the write and a
+    // foreground sender's transient auth blip becomes a producer-fatal terminal -- the
+    // exact failure this policy exists to prevent.
     private volatile boolean hasEverConnected;
     // Cause of the outage the reconnect loop is currently riding out, or null once a
     // connect succeeds. Written by the I/O thread, read by the producer thread so a

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

@@ -566,47 +566,6 @@ public long tryAppend(long payloadAddr, int payloadLen) {
         return offset;
     }
 
-    /**
-     * Walks every published frame in this segment and returns the FSN of the
-     * LAST frame whose payload does NOT carry the given flag bit, or {@code -1}
-     * when every frame carries it (or the segment is empty).
-     * <p>
-     * A frame counts as carrying the flag ONLY when it positively parses as a
-     * message of the expected protocol: payload at least {@code minPayloadLen}
-     * bytes AND the little-endian u32 at payload offset 0 equals
-     * {@code headerMagic} AND the byte at {@code flagsOffset} has
-     * {@code flagMask} set. Anything else -- short frames, foreign payloads,
-     * magic mismatches -- counts as NOT carrying the flag. This direction is
-     * deliberate: the caller retires (trims) frames ABOVE the returned FSN,
-     * so a frame we cannot positively identify must act as a retirement
-     * barrier, never as trimmable. Misclassifying an unknown frame as
-     * deferred would silently discard data that should replay.
-     * <p>
-     * Producer-thread only, and only meaningful before new appends race the
-     * walk (recovery time). Used to locate the last commit-bearing QWP frame
-     * below a potentially orphaned FLAG_DEFER_COMMIT tail: frames above the
-     * returned FSN all carry the flag, i.e. they belong to a transaction whose
-     * commit frame was never published.
-     */
-    public long findLastFrameFsnWithoutPayloadFlag(int flagsOffset, int flagMask, int headerMagic, int minPayloadLen) {
-        long best = -1L;
-        long off = HEADER_SIZE;
-        long frames = frameCount;
-        for (long i = 0; i < frames; i++) {
-            int payloadLen = Unsafe.getUnsafe().getInt(mmapAddress + off + 4);
-            long payload = mmapAddress + off + FRAME_HEADER_SIZE;
-            boolean flagSet = payloadLen >= minPayloadLen
-                    && payloadLen > flagsOffset
-                    && Unsafe.getUnsafe().getInt(payload) == headerMagic
-                    && (Unsafe.getUnsafe().getByte(payload + flagsOffset) & flagMask) != 0;
-            if (!flagSet) {
-                best = baseSeq + i;
-            }
-            off += FRAME_HEADER_SIZE + payloadLen;
-        }
-        return best;
-    }
-
     /**
      * Feeds every recovered frame in this segment to the engine's single-pass
      * recovery fold. Segments are visited oldest-first by {@link SegmentRing}.

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

@@ -172,6 +172,11 @@ public final class PersistedSymbolDict implements QuietCloseable {
     // tests inject a fault facade to exercise recovery I/O failures (a truncate
     // that cannot drop a torn tail, a short write) without a real broken disk.
     private final FilesFacade ff;
+    // Slot-qualified path, retained purely so diagnostics can name WHICH dictionary they
+    // are about: one JVM routinely holds many (a foreground sender plus N orphan
+    // drainers), so a warning carrying only the bare filename is unattributable exactly
+    // when it fires.
+    private final String filePath;
     // Production writes directly into segmented append mappings. Wrapping facades retain the
     // positioned-write path by default so fault tests can inject short writes through ff.write;
     // mmap-specific fault facades opt in through FilesFacade.isMmapAllowed().
@@ -209,6 +214,7 @@ public final class PersistedSymbolDict implements QuietCloseable {
 
     private PersistedSymbolDict(
             FilesFacade ff,
+            String filePath,
             int fd,
             long appendOffset,
             int size,
@@ -217,6 +223,7 @@ private PersistedSymbolDict(
             boolean mappedRecoveryInput
     ) {
         this.ff = ff;
+        this.filePath = filePath;
         this.fd = fd;
         this.mappedAppend = ff.isMmapAllowed();
         this.mappedRecoveryInput = mappedRecoveryInput;
@@ -510,9 +517,13 @@ public synchronized void close() {
                 Unsafe.free(loadedEntriesAddr, loadedEntriesLen, MemoryTag.NATIVE_DEFAULT);
             } catch (Throwable ignored) {
             }
-            // Null after freeing (like scratchAddr below) so a future accessor that
-            // reads loadedEntriesAddr()/loadedEntriesLen() post-close cannot
-            // dereference freed native memory; the getters are not closed-guarded.
+            // Null after freeing (like scratchAddr below) so a CONSTRUCTION-PHASE reader
+            // that runs after close observes 0 rather than a dangling pointer. This is
+            // NOT a cross-thread guard, and the getters' own javadoc is the accurate
+            // statement: they are unsynchronised reads of non-volatile fields, so a
+            // caller on another thread has no guarantee of seeing this write and can
+            // still dereference freed memory. Ordering, not nulling, is what makes the
+            // borrow safe -- every owner closes its loop before the engine.
             loadedEntriesAddr = 0L;
             loadedEntriesLen = 0;
         }
@@ -533,7 +544,7 @@ public synchronized void close() {
                 if (!ff.truncate(fd, appendOffset)) {
                     LOG.warn("symbol dict {} could not trim mmap reserve to {}; recovery will "
                                     + "discard the zero-filled tail on the next open",
-                            FILE_NAME, appendOffset);
+                            filePath, appendOffset);
                 }
             } catch (Throwable ignored) {
             }
@@ -753,7 +764,7 @@ private static PersistedSymbolDict openExisting(FilesFacade ff, String filePath,
                 throw new IllegalStateException("could not drop torn/stale symbol dictionary tail");
             }
             return new PersistedSymbolDict(
-                    ff, fd, scan.validLen, scan.count, entriesAddr, entriesLen, mappedInput);
+                    ff, filePath, fd, scan.validLen, scan.count, entriesAddr, entriesLen, mappedInput);
         } catch (Throwable t) {
             if (inputAddr != 0L) {
                 if (mappedInput) {
@@ -921,7 +932,7 @@ private static PersistedSymbolDict openFresh(FilesFacade ff, String filePath) {
                 Unsafe.free(hdr, HEADER_SIZE, MemoryTag.NATIVE_DEFAULT);
             }
         }
-        return new PersistedSymbolDict(ff, fd, HEADER_SIZE, 0, 0L, 0, false);
+        return new PersistedSymbolDict(ff, filePath, fd, HEADER_SIZE, 0, 0L, 0, false);
     }
 
     /**

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

@@ -87,7 +87,7 @@ void accept(long fsn, long payload, int payloadLen) {
 
         // Only a positively identified deferred QWP frame can belong to an
         // uncommitted tail. Short, foreign or otherwise non-QWP payloads remain
-        // retirement barriers, matching findLastFrameFsnWithoutPayloadFlag.
+        // retirement barriers.
         if (!isQwp || (flags & QwpConstants.FLAG_DEFER_COMMIT) == 0) {
             committedBoundaryFsn = fsn;
             committedCoverage = runningCoverage;
@@ -159,8 +159,6 @@ void finish() {
             committedRawCount = 0;
             committedGap = true;
         }
-        runningRawLen = committedRawLen;
-        runningRawCount = committedRawCount;
         if (rawCapacity == committedRawLen) {
             return;
         }

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

@@ -656,35 +656,6 @@ public synchronized MmapSegment firstSealed() {
     }
 
     /** Active segment -- exposed for the I/O thread's "send next batch" path. */
-    /**
-     * Walks every published frame in the ring (sealed segments plus the active
-     * segment) and returns the FSN of the LAST frame whose payload does NOT
-     * carry the given flag bit, or {@code -1} when every published frame
-     * carries it (or the ring is empty). All frames above the returned FSN
-     * carry the flag.
-     * <p>
-     * Recovery-time helper: locates the last commit-bearing QWP frame below a
-     * potentially orphaned FLAG_DEFER_COMMIT tail left behind by a producer
-     * that crashed (or closed) mid-transaction. Call before the I/O loop and
-     * producer start appending; the walk is not synchronized against appends
-     * into the active segment. See
-     * {@link MmapSegment#findLastFrameFsnWithoutPayloadFlag} for the
-     * positive-identification contract: frames that do not parse as protocol
-     * messages count as commit-bearing (retirement barriers), never as
-     * trimmable.
-     */
-    public synchronized long findLastFsnWithoutPayloadFlag(int flagsOffset, int flagMask, int headerMagic, int minPayloadLen) {
-        long best = -1L;
-        for (int i = sealedHead, n = sealedSegments.size(); i < n; i++) {
-            long fsn = sealedSegments.get(i).findLastFrameFsnWithoutPayloadFlag(flagsOffset, flagMask, headerMagic, minPayloadLen);
-            if (fsn > best) {
-                best = fsn;
-            }
-        }
-        long fsn = active.findLastFrameFsnWithoutPayloadFlag(flagsOffset, flagMask, headerMagic, minPayloadLen);
-        return Math.max(best, fsn);
-    }
-
     /**
      * Performs the one ordered recovery fold across sealed segments and the
      * active segment. The returned native suffix remains owned by the caller.