Close the remaining four correctness holes in SF recovery

C5. The foreground send loop CONSUMED the engine's recovery sources --
pd.takeLoadedEntries() plus engine.releaseRecoveredSymbolStorage() --
making the hand-off one-shot while leaving a second construction
reachable: ensureConnected's catch closes and nulls cursorSendLoop
precisely so a caller can retry, and `connected` is set only at the very
end. Neither call cleared the counts that gate re-seeding, so the retry
saw recoveredSize() > 0 with loadedEntriesLen() == 0, left sentDictCount
at 0, and then either tripped the baseline mismatch or latched the
torn-dictionary terminal on a healthy slot. Both policies now BORROW, as
the orphan drainer already did; the engine frees at close. Lifetime holds
in both directions because QwpWebSocketSender.close() and
BackgroundDrainer's finally both close the loop before the engine. The
two transfer methods had no other caller and are deleted.

C6. A pre-21 unsafe-access fault is delivered asynchronously, so one
taken while scanning a segment can surface at MmapSegment.openExisting's
RETURN -- in SegmentRing's frame, before `seg = openExisting(...)` ever
assigns. The segment was fully constructed, so its fd and whole-file
mapping leaked with nothing referencing them, while the log claimed a
skip that had not happened. openExisting now publishes the segment into
a caller-supplied holder immediately before returning, and SegmentRing's
finally closes whichever handle survived. The holder is cleared at every
ownership transfer, so it can never close a segment `opened` now holds.

C7. accept() checkpoints committedRawLen/Count only at a commit-bearing
frame, while foldDelta's gap-reset rewinds runningRawLen/Count to 0
without touching them -- so the next appendRaw overwrites the committed
bytes in place. Harmless when a commit-bearing frame follows, but a reset
inside an uncommitted deferred tail never gets that refresh and finish()
then published the old counts over the tail's bytes: recovery decoded the
tail's symbol under the committed id while the frame that actually
replays registered a different one. finish() now detects the rewind and
fails clean instead.

C8. markFailed is best-effort and returns silently when it cannot open
the sentinel -- and a full or read-only disk, the condition that makes a
slot unreplayable, is exactly the condition that makes writing it fail.
Twelve lines after quarantining, the same build() dispatches its orphan
drainers, and the quarantined copy was still a candidate: unbounded churn
against data explicitly declared human-in-the-loop, repeating on every
restart. OrphanScanner now excludes quarantined slots BY NAME, and owns
the reserved infix that Sender.build() renames with.

Regression tests, each verified to fail with its fix reverted:
testForegroundLoopReSeedsAfterAClosedFirstAttempt (UnreplayableSlot
"recovery symbol baseline mismatch [expected=2, actual=0]"),
testGapResetInsideTheDeferredTailFailsCleanInsteadOfMisattributing
(expected:<-1> but was:<1>), and
testQuarantinedSlotIsNotACandidateEvenWithoutTheFailedSentinel. The two
MirrorLeakTest cases that pinned the old ownership transfer now pin
borrowing, including that closing a borrowed loop leaves the engine's
buffer alive.

C6 has no regression test: reproducing it needs a fault delivered at one
specific instruction boundary on a pre-21 JIT, which is the same
JIT-dependent delivery the existing MmapSegmentRecoveryFaultTest guards
cannot pin either. The change is a strict improvement over having no
handle at all, and the existing multi-segment recovery tests cover the
double-close risk it introduces.

Full client suite: 2693 tests green.

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

Author: glasstiger

core/src/main/java/io/questdb/client/Sender.java

@@ -1026,7 +1026,8 @@ final class LineSenderBuilder {
         // sender's own slot name, so a restarted sender does not re-adopt it as its own;
         // quarantineTornSlot then marks it .failed, so the orphan drainer skips it too and
         // the bytes stay put for a human to inspect and resend.
-        private static final String QUARANTINE_SLOT_SUFFIX = ".unreplayable-";
+        private static final String QUARANTINE_SLOT_SUFFIX =
+                OrphanScanner.QUARANTINE_SLOT_INFIX;
         private final ObjList<String> hosts = new ObjList<>();
         private final IntList ports = new IntList();
         private long authTimeoutMillis = QwpWebSocketSender.DEFAULT_AUTH_TIMEOUT_MS;

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

@@ -809,12 +809,6 @@ void copyRecoveredSymbolSuffix(int baseline, long target) {
         }
     }
 
-    void releaseRecoveredSymbolStorage() {
-        if (recoveredFrameAnalysis != null) {
-            recoveredFrameAnalysis.releaseRawStorage();
-        }
-    }
-
     @TestOnly
     public long recoveryFramesVisited() {
         return recoveredFrameAnalysis == null ? 0L : recoveredFrameAnalysis.framesVisited();

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

@@ -725,7 +725,6 @@ public CursorWebSocketSendLoop(WebSocketClient client, CursorSendEngine engine,
         // tracks. When the dictionary could not be opened, pd is null, the mirror
         // simply starts empty and grows from the frames themselves.
         PersistedSymbolDict pd = engine.getPersistedSymbolDict();
-        int persistedPrefixLen = 0;
         if (pd != null && pd.recoveredSize() > 0) {
             int len = pd.loadedEntriesLen();
             if (len > 0) {
@@ -734,7 +733,6 @@ public CursorWebSocketSendLoop(WebSocketClient client, CursorSendEngine engine,
                 // one engine can create several sessions during capability-gap
                 // recycling. A borrowed mirror performs copy-on-write if a recovered
                 // frame suffix must extend it.
-                persistedPrefixLen = len;
                 sentDictBytesAddr = pd.loadedEntriesAddr();
                 sentDictBytesCapacity = len;
                 sentDictBytesLen = len;
@@ -803,24 +801,24 @@ public CursorWebSocketSendLoop(WebSocketClient client, CursorSendEngine engine,
             releaseSentDictBytes();
             throw t;
         }
-        // QwpWebSocketSender seeds the producer dictionary before constructing
-        // its one foreground loop, so after the loop has built its mirror it can
-        // retire both engine-owned recovery sources. BackgroundDrainer must retain
-        // them for later recycled loops and therefore keeps borrowing them.
-        if (reconnectPolicy == ReconnectPolicy.FOREGROUND) {
-            if (persistedPrefixLen > 0) {
-                long persistedPrefixAddr = pd.takeLoadedEntries();
-                if (sentDictBytesOwned) {
-                    // Copy-on-grow already produced the combined mirror. Retire the
-                    // no-longer-needed persisted prefix after successful construction.
-                    Unsafe.free(persistedPrefixAddr, persistedPrefixLen, MemoryTag.NATIVE_DEFAULT);
-                } else {
-                    assert persistedPrefixAddr == sentDictBytesAddr;
-                    sentDictBytesOwned = true;
-                }
-            }
-            engine.releaseRecoveredSymbolStorage();
-        }
+        // BOTH policies borrow the engine-owned recovery sources; neither consumes them.
+        //
+        // The foreground path used to take ownership here -- pd.takeLoadedEntries() plus
+        // engine.releaseRecoveredSymbolStorage() -- which made the hand-off ONE-SHOT while
+        // leaving a second construction reachable: QwpWebSocketSender.ensureConnected's
+        // catch closes and nulls cursorSendLoop precisely so a caller can retry, and
+        // `connected` is only set at the very end. takeLoadedEntries zeroes addr/len but
+        // NOT size, and releaseRecoveredSymbolStorage zeroes the raw buffer but NOT
+        // baseline, so the retry saw pd.recoveredSize() > 0 with loadedEntriesLen() == 0,
+        // left sentDictCount at 0, and then either tripped checkedRecoveryAnalysis'
+        // baseline mismatch or -- when recoveredMaxSymbolDeltaStart == 0 -- latched the
+        // torn-dictionary terminal ("resend required") on a perfectly healthy slot.
+        //
+        // Borrowing removes the state machine instead of trying to make it idempotent.
+        // Lifetime is safe in both directions: QwpWebSocketSender.close() closes the loop
+        // before the engine, and BackgroundDrainer's finally does the same, so the buffers
+        // always outlive their borrower. Any growth copy-on-writes into loop-owned memory
+        // (ensureSentDictCapacity), and releaseSentDictBytes frees only what the loop owns.
         this.fsnAtZero = fsnAtZero;
         this.parkNanos = parkNanos;
         this.reconnectFactory = reconnectFactory;

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

@@ -335,6 +335,23 @@ public static MmapSegment openExisting(String path) {
      * {@link FilesFacade#length(String)}.
      */
     public static MmapSegment openExisting(FilesFacade ff, String path) {
+        return openExisting(ff, path, null);
+    }
+
+    /**
+     * As {@link #openExisting(FilesFacade, String)}, but also publishes the constructed
+     * segment into {@code inFlight[0]} immediately before returning it.
+     * <p>
+     * Pre-JDK-21 HotSpot delivers an unsafe-access fault ASYNCHRONOUSLY -- at the next
+     * return or safepoint check rather than at the faulting instruction (JDK-8283699).
+     * A fault taken while scanning this segment can therefore surface at this method's
+     * RETURN, in the caller's frame, so the caller's {@code seg = openExisting(...)}
+     * assignment never happens even though the fd and the whole-file mapping were
+     * already acquired -- and nothing is left holding them. The holder gives the
+     * caller's {@code finally} something to close. Callers MUST clear {@code inFlight[0]}
+     * wherever they transfer ownership, or they will close a segment they just handed on.
+     */
+    static MmapSegment openExisting(FilesFacade ff, String path, MmapSegment[] inFlight) {
         long fileSize = ff.length(path);
         if (fileSize < HEADER_SIZE) {
             throw new MmapSegmentException("file shorter than header: " + path + " size=" + fileSize);
@@ -382,8 +399,12 @@ public static MmapSegment openExisting(FilesFacade ff, String path) {
                                 + "Investigate disk health or unexpected writer crash.",
                         path, tornTail, lastGood, fileSize, count);
             }
-            return new MmapSegment(path, fd, addr, fileSize, baseSeq,
+            MmapSegment segment = new MmapSegment(path, fd, addr, fileSize, baseSeq,
                     lastGood, count, false, tornTail, version);
+            if (inFlight != null) {
+                inFlight[0] = segment;
+            }
+            return segment;
         } catch (Throwable t) {
             if (addr != Files.FAILED_MMAP_ADDRESS) {
                 Files.munmap(addr, fileSize, MemoryTag.MMAP_DEFAULT);

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

@@ -61,6 +61,20 @@ public final class OrphanScanner {
 
     /** Name of the sentinel that disqualifies a slot from auto-drain. */
     public static final String FAILED_SENTINEL_NAME = ".failed";
+    /**
+     * Reserved infix for a slot {@code Sender.build()} set aside as unreplayable. Such a
+     * copy is kept for forensics and a manual resend; it is NEVER drainable, so the
+     * scanner excludes it BY NAME.
+     * <p>
+     * Excluding it by name rather than by the {@code .failed} sentinel alone is what makes
+     * the exclusion reliable. {@code markFailed} is best-effort and returns silently when
+     * it cannot open the sentinel -- and the condition that makes a slot unreplayable (a
+     * full or read-only disk) is exactly the condition that makes writing the sentinel
+     * fail. Without the name check, the SAME build() that quarantined the slot dispatches
+     * its orphan drainers twelve lines later and re-adopts it: unbounded churn against
+     * data explicitly declared human-in-the-loop, repeating on every restart.
+     */
+    public static final String QUARANTINE_SLOT_INFIX = ".unreplayable-";
 
     private OrphanScanner() {
     }
@@ -271,12 +285,29 @@ public static boolean isCandidateOrphan(String slotPath) {
         if (!Files.exists(slotPath)) {
             return false;
         }
+        if (isQuarantinedSlotName(slotPath)) {
+            return false;
+        }
         if (Files.exists(slotPath + "/" + FAILED_SENTINEL_NAME)) {
             return false;
         }
         return hasAnySegmentFile(slotPath);
     }
 
+    /**
+     * Whether {@code slotPath}'s last path element names a quarantined slot. Independent
+     * of the {@code .failed} sentinel, which is best-effort -- see
+     * {@link #QUARANTINE_SLOT_INFIX}.
+     */
+    public static boolean isQuarantinedSlotName(String slotPath) {
+        if (slotPath == null) {
+            return false;
+        }
+        int slash = Math.max(slotPath.lastIndexOf('/'), slotPath.lastIndexOf('\\'));
+        String name = slash < 0 ? slotPath : slotPath.substring(slash + 1);
+        return name.contains(QUARANTINE_SLOT_INFIX);
+    }
+
     /**
      * Drops a {@link #FAILED_SENTINEL_NAME} file in {@code slotPath}.
      * Idempotent — touching an existing sentinel is a no-op (its presence

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

@@ -567,18 +567,6 @@ public int loadedEntriesLen() {
         return loadedEntriesLen;
     }
 
-    /**
-     * Transfers the recovered entry buffer to the foreground send loop. The
-     * symbol count remains unchanged because it is also the append baseline;
-     * only the native-buffer ownership moves. Construction-phase only.
-     */
-    synchronized long takeLoadedEntries() {
-        long addr = loadedEntriesAddr;
-        loadedEntriesAddr = 0L;
-        loadedEntriesLen = 0;
-        return addr;
-    }
-
     /**
      * Number of symbols {@link #open} recovered from disk -- the exact entry count of
      * {@link #loadedEntriesAddr()} / {@link #loadedEntriesLen()}. Unlike {@link #size()}

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

@@ -51,6 +51,10 @@ final class RecoveredFrameAnalysis implements QuietCloseable {
     private int committedRawCount;
     private int committedRawLen;
     private long framesVisited;
+    // Set when a gap-reset rewinds the raw write cursor, cleared at every commit
+    // checkpoint. Still set at finish() => the committed snapshot's bytes were
+    // overwritten by an uncommitted tail and must not be published. See finish().
+    private boolean hasRewoundSinceCommit;
     private boolean runningGap;
     private boolean runningUnackedGap;
     private long runningMaxDeltaEnd;
@@ -92,6 +96,7 @@ void accept(long fsn, long payload, int payloadLen) {
             committedMaxDeltaStart = runningMaxDeltaStart;
             committedRawLen = runningRawLen;
             committedRawCount = runningRawCount;
+            hasRewoundSinceCommit = false;
         }
     }
 
@@ -142,6 +147,18 @@ long framesVisited() {
      * deferred tail. Call exactly once after the ordered recovery walk.
      */
     void finish() {
+        if (hasRewoundSinceCommit) {
+            // A gap-reset rewound the write cursor after the last commit checkpoint, so
+            // the deferred tail has overwritten the bytes committedRawLen/Count still
+            // describe. Publishing them would decode the TAIL's entries under the
+            // committed ids -- exactly the silent misattribution this analysis exists to
+            // prevent. Fail clean instead: no suffix, and a gap so coverage() reports -1
+            // and the producer falls back to the persisted prefix (or, when unacked
+            // frames depend on the missing ids, quarantines).
+            committedRawLen = 0;
+            committedRawCount = 0;
+            committedGap = true;
+        }
         runningRawLen = committedRawLen;
         runningRawCount = committedRawCount;
         if (rawCapacity == committedRawLen) {
@@ -274,6 +291,13 @@ private void foldDelta(long fsn, long p, long limit) {
             }
             runningGap = false;
             runningCoverage = baseline;
+            // Rewinding the write cursor to 0 means the NEXT appendRaw overwrites
+            // [0, committedRawLen) in place -- bytes the committed snapshot still
+            // counts. That is harmless when a commit-bearing frame follows (accept()
+            // re-checkpoints and clears this flag), but a reset inside an uncommitted
+            // deferred TAIL never gets that refresh, and finish() would then hand out
+            // the old counts over the tail's bytes: silent symbol misattribution.
+            hasRewoundSinceCommit = true;
             runningRawLen = 0;
             runningRawCount = 0;
         }

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

@@ -205,15 +205,20 @@ public static SegmentRing openExisting(FilesFacade ff, String sfDir, long maxByt
         try {
             try {
                 int rc = 1;
+                // Reused across files: MmapSegment.openExisting publishes the segment
+                // here before returning, so a pre-21 late-delivered unsafe-access fault
+                // that steals the return value still leaves us something to close.
+                final MmapSegment[] inFlight = new MmapSegment[1];
                 while (rc > 0) {
                     String name = Files.utf8ToString(Files.findName(find));
                     if (name != null
                             && name.endsWith(".sfa")
                             && !isLegacyReaderGuard(name)) {
                         String path = sfDir + "/" + name;
                         MmapSegment seg = null;
+                        inFlight[0] = null;
                         try {
-                            seg = MmapSegment.openExisting(ff, path);
+                            seg = MmapSegment.openExisting(ff, path, inFlight);
                             // Filter out empty leftovers -- typically hot-spare
                             // segments the manager pre-allocated for a prior
                             // session that never got rotated into active. They
@@ -238,6 +243,7 @@ public static SegmentRing openExisting(FilesFacade ff, String sfDir, long maxByt
                                 long torn = seg.tornTailBytes();
                                 seg.close();
                                 seg = null;
+                                inFlight[0] = null;
                                 if (torn > 0) {
                                     Files.rename(path, path + ".corrupt");
                                 } else {
@@ -246,6 +252,7 @@ public static SegmentRing openExisting(FilesFacade ff, String sfDir, long maxByt
                             } else {
                                 opened.add(seg);
                                 seg = null;
+                                inFlight[0] = null;
                             }
                         } catch (Throwable t) {
                             // Per-file data error (bad magic, bad header,
@@ -285,9 +292,16 @@ public static SegmentRing openExisting(FilesFacade ff, String sfDir, long maxByt
                             // open and transfer -- most importantly an OOM
                             // from ObjList.add growing its backing array
                             // after the mmap+fd were already acquired.
-                            if (seg != null) {
+                            // seg may still be null while the segment was fully
+                            // constructed: a late-delivered mmap fault can surface at
+                            // openExisting's RETURN, before the assignment. inFlight is
+                            // cleared at every ownership transfer above, so this can
+                            // never close a segment `opened` now holds.
+                            MmapSegment orphan = seg != null ? seg : inFlight[0];
+                            inFlight[0] = null;
+                            if (orphan != null) {
                                 try {
-                                    seg.close();
+                                    orphan.close();
                                 } catch (Throwable closeErr) {
                                     LOG.warn("openExisting: error closing in-flight segment {}",
                                             path, closeErr);