Fix SF recovery, lock and quarantine safety bugs

Address blocking review findings in the store-and-forward paths.

SegmentRing.openExisting no longer detaches the active segment from
`opened` before the ring owns it: `new SegmentRing` and the sealed-add
loop can throw (ObjList growth, native OOM), and the catch iterated only
`opened`, leaking the active segment's mapping and fd. Ownership now
transfers in one step and `opened` is cleared last.

PersistedSymbolDict.open gains the in-flight holder MmapSegment already
uses. Recovery reads the file through a mapping, and pre-JDK-21 HotSpot
delivers an unsafe-access fault at the method's return, past the inner
catch; without a holder the built instance's fd and buffer leak and the
fault escapes build() instead of degrading to full-dictionary frames.

commitMappedChunk checksums the append mapping with Crc32c.updateUnsafe
rather than the native update. ff.allocate can leave the window over
uncommitted blocks (ENOSPC, quota, a crash-torn tail); a SIGBUS there
aborts the JVM through JNI but is a catchable InternalError at an Unsafe
site, matching scanFrames.

SlotLock.removeOrphanLogical acquires the lock before unlinking it.
Removing a file another party holds frees the pathname without releasing
the flock, so the next acquireLogical locks a second inode -- two owners
of a mutual-exclusion lock. QwpWebSocketSender.close carries a
reclaim-logical-lock flag that connect()'s rollback clears, so a failed
connect no longer unlinks the lock Sender.build() still holds.

Sender.build reports an abandoned quarantined slot through the configured
errorHandler, not only via LOG.error: the client ships slf4j-api with no
binding, so an embedder without a provider would learn of the loss
nowhere.

Files.DIR_MODE_SHARED's javadoc no longer claims the sticky bit survives:
Linux mkdir(2) masks it off, so the restricted-deletion guarantee is
best-effort and lock safety rests on the acquire-before-unlink above.

SlotLockTest pins the held-lock case both ways.

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

Author: glasstiger

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

@@ -1658,7 +1658,7 @@ public Sender build() {
                             quarantined = true;
                             cursorEngine = quarantineTornSlot(
                                     cursorEngine, e, sfDir, senderId, slotPath, actualSfMaxBytes,
-                                    actualSfMaxTotalBytes, actualSfAppendDeadlineNanos);
+                                    actualSfMaxTotalBytes, actualSfAppendDeadlineNanos, errorHandler);
                         } catch (Throwable t) {
                             // connect() failed before ownership of cursorEngine
                             // transferred — close it ourselves. close(false)
@@ -3065,7 +3065,8 @@ private static long parseSizeValue(@NotNull StringSink value, @NotNull String na
         private static CursorSendEngine quarantineTornSlot(
                 CursorSendEngine torn, UnreplayableSlotException cause, String sfDir,
                 String senderId, String slotPath,
-                long sfMaxBytes, long sfMaxTotalBytes, long sfAppendDeadlineNanos
+                long sfMaxBytes, long sfMaxTotalBytes, long sfAppendDeadlineNanos,
+                io.questdb.client.SenderErrorHandler errorHandler
         ) {
             // The verdict, and the reason, come from the recovery seed -- the only code that
             // has tried every source of truth. Recomputing them here would mean a second,
@@ -3106,6 +3107,34 @@ private static CursorSendEngine quarantineTornSlot(
             LOG.error("{} -- the slot has been set aside at {} and the affected data must be resent; "
                             + "this sender continues on a fresh, empty slot at {}",
                     detail, quarantinePath, slotPath);
+            // build() no longer throws for this -- it starts the producer on a fresh slot so
+            // the outage stays bounded. But abandoning buffered rows is precisely the event a
+            // caller must be able to act on, and LOG.error alone cannot carry it: this client
+            // ships slf4j-api with no binding, so an embedding app with no provider gets a NOP
+            // logger and the loss is announced nowhere. Deliver it programmatically too, so an
+            // errorHandler can alert / page / record it. Dispatched synchronously here because
+            // the async SenderErrorDispatcher belongs to the connected sender, which does not
+            // exist yet at build time. A throwing handler must not turn a contained outage back
+            // into a failed build, so swallow anything it raises.
+            if (errorHandler != null) {
+                try {
+                    errorHandler.onError(new SenderError(
+                            SenderError.Category.PROTOCOL_VIOLATION,
+                            SenderError.Policy.TERMINAL,
+                            SenderError.NO_STATUS_BYTE,
+                            detail + " [slot set aside at " + quarantinePath
+                                    + "; sender continues on a fresh slot at " + slotPath + ']',
+                            SenderError.NO_MESSAGE_SEQUENCE,
+                            SenderError.NO_MESSAGE_SEQUENCE,
+                            SenderError.NO_MESSAGE_SEQUENCE,
+                            null,
+                            System.nanoTime()
+                    ));
+                } catch (Throwable handlerFailure) {
+                    LOG.error("sender error handler threw while reporting a quarantined slot: {}",
+                            String.valueOf(handlerFailure));
+                }
+            }
             return new CursorSendEngine(slotPath, sfMaxBytes, sfMaxTotalBytes, sfAppendDeadlineNanos);
         }
 

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

@@ -306,6 +306,17 @@ public class QwpWebSocketSender implements Sender {
     //         contained in that loop and never reach the producer.
     private Sender.InitialConnectMode initialConnectMode = Sender.InitialConnectMode.OFF;
     private boolean ownsCursorEngine;
+    // Whether close() may let the engine reclaim the parent-anchored LOGICAL slot lock.
+    // False only while an outer frame holds it: Sender.build() acquires it for the whole
+    // construct -> connect -> quarantine transition, and connect()'s own rollback closes
+    // this sender from INSIDE that scope. A fresh slot is "fully drained" by definition
+    // (publishedFsn < 0), so the default close(true) took the reclaim branch and unlinked
+    // the very lock file build() was holding -- on POSIX that frees the pathname without
+    // releasing the flock, so the next acquireLogical creates a SECOND inode and locks it.
+    // build()'s own careful close(false) calls could not prevent it, because connect()
+    // closed the engine first. Reset to true once connect() hands ownership back, so a
+    // later user-initiated close() still retires the lock normally.
+    private boolean reclaimLogicalSlotLockOnClose = true;
     private long pendingBytes;
     // Set true by close() once the SF slot flock has been released (the normal
     // teardown path). Stays false if close() bailed early with the I/O thread
@@ -821,6 +832,10 @@ public static QwpWebSocketSender connect(
             // handler -- and close() accumulates cleanup errors and ends in
             // rethrowTerminal, so letting a close failure propagate here would REPLACE t
             // and silently demote a recoverable slot back to the permanent build() brick.
+            // This rollback always runs inside Sender.build()'s acquireLogical scope, so
+            // the logical slot lock is held one frame up. Closing the engine with the
+            // default reclaim would unlink the lock file build() is still holding.
+            sender.reclaimLogicalSlotLockOnClose = false;
             try {
                 sender.close();
             } catch (Throwable closeFailure) {
@@ -1283,7 +1298,7 @@ public void close() {
                 if (ownsCursorEngine && cursorEngine != null && cursorSendLoop != null
                         && !cursorSendLoop.delegateEngineClose()) {
                     try {
-                        cursorEngine.close();
+                        cursorEngine.close(reclaimLogicalSlotLockOnClose);
                     } catch (Throwable t) {
                         LOG.error("Error closing owned CursorSendEngine: {}", String.valueOf(t));
                         terminalError = captureCloseError(terminalError, t);
@@ -1325,7 +1340,7 @@ public void close() {
 
             if (ownsCursorEngine && cursorEngine != null) {
                 try {
-                    cursorEngine.close();
+                    cursorEngine.close(reclaimLogicalSlotLockOnClose);
                 } catch (Throwable t) {
                     LOG.error("Error closing owned CursorSendEngine: {}", String.valueOf(t));
                     terminalError = captureCloseError(terminalError, t);

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

@@ -286,7 +286,32 @@ public static PersistedSymbolDict open(FilesFacade ff, String slotDir) {
             // truncates, and these bytes are the only copy of state the surviving
             // delta frames reference. A null degrades this slot to full
             // self-sufficient frames and preserves the file for a later attempt.
-            return openExisting(ff, filePath, existing);
+            //
+            // The holder catches what openExisting's own catch structurally cannot.
+            // Recovery reads the file through a mapping (Unsafe loads in
+            // scanAndCopyRecoveredChunks), and pre-JDK-21 HotSpot delivers an
+            // unsafe-access fault ASYNCHRONOUSLY -- at the next return or safepoint,
+            // which can be openExisting's own return, in THIS frame. The instance is
+            // fully built by then and owns an fd plus a buffer as large as the file,
+            // but the assignment never happens, so nothing else can release them.
+            // MmapSegment.openExisting takes an inFlight[] for exactly this shape; the
+            // dictionary is at least as exposed, because ensureAppendMap grows the file
+            // with ff.allocate and the reserve truncate is skipped after a crash, so a
+            // sparse tail is routine. Without this, the fault also escapes open()
+            // entirely -- past CursorSendEngine's constructor and out of Sender.build()
+            // -- turning the documented "degrade to full-dictionary frames" into a
+            // sender that cannot be constructed at all.
+            PersistedSymbolDict[] inFlight = new PersistedSymbolDict[1];
+            try {
+                return openExisting(ff, filePath, existing, inFlight);
+            } catch (Throwable t) {
+                if (inFlight[0] != null) {
+                    inFlight[0].close();
+                }
+                LOG.warn("symbol dict {} recovery faulted late; falling back to "
+                        + "full-dictionary frames (file left intact)", filePath, t);
+                return null;
+            }
         }
         // Absent, or a sub-header stub left by a crash inside openFresh: no
         // load-bearing content to lose, so create it.
@@ -693,7 +718,15 @@ public int size() {
         return size;
     }
 
-    private static PersistedSymbolDict openExisting(FilesFacade ff, String filePath, long fileLen) {
+    /**
+     * @param inFlight single-slot holder the caller must close when this method throws.
+     *                 Set immediately before the return, so a late-delivered mmap access
+     *                 fault -- which pre-JDK-21 HotSpot may raise at this method's RETURN,
+     *                 in the caller's frame, past the catch below -- still leaves the fully
+     *                 built instance (fd plus loaded-entry buffer) reachable by someone.
+     */
+    private static PersistedSymbolDict openExisting(
+            FilesFacade ff, String filePath, long fileLen, PersistedSymbolDict[] inFlight) {
         int fd = ff.openRW(filePath);
         if (fd < 0) {
             LOG.warn("symbol dict {} could not be opened (rc={}); "
@@ -763,8 +796,12 @@ private static PersistedSymbolDict openExisting(FilesFacade ff, String filePath,
             if (scan.validLen < len && !ff.truncate(fd, scan.validLen)) {
                 throw new IllegalStateException("could not drop torn/stale symbol dictionary tail");
             }
-            return new PersistedSymbolDict(
+            PersistedSymbolDict dict = new PersistedSymbolDict(
                     ff, filePath, fd, scan.validLen, scan.count, entriesAddr, entriesLen, mappedInput);
+            // Publish to the holder BEFORE returning: from here on the caller owns the
+            // cleanup, including when the return itself faults. See the parameter doc.
+            inFlight[0] = dict;
+            return dict;
         } catch (Throwable t) {
             if (inputAddr != 0L) {
                 if (mappedInput) {
@@ -995,9 +1032,17 @@ private long checkedRequiredOffset(long recordLen) {
     private void commitMappedChunk(long recStart, int hdrLen, int entriesLen, int count) {
         long bodyLen = (long) hdrLen + entriesLen;
         long recLen = bodyLen + CRC_SIZE;
+        // updateUnsafe, NOT the native update: this checksums bytes inside the append
+        // MAPPING. ensureAppendMap grows the file with ff.allocate, whose native fallback
+        // is ftruncate, so the window can cover blocks the filesystem has not committed
+        // (ENOSPC, a quota, a sparse tail left by a crash) -- and touching one raises
+        // SIGBUS. Inside JNI that ABORTS THE JVM with no recovery; at an Unsafe intrinsic
+        // site HotSpot converts it to a catchable InternalError. MmapSegment.scanFrames
+        // already uses updateUnsafe over the same class of mapping for exactly this
+        // reason. Costs nothing measurable and drops a JNI transition per flush.
         Unsafe.getUnsafe().putInt(
                 recStart + bodyLen,
-                Crc32c.update(Crc32c.INIT, recStart, bodyLen));
+                Crc32c.updateUnsafe(Crc32c.INIT, recStart, bodyLen));
         Unsafe.getUnsafe().storeFence();
         appendOffset += recLen;
         size += count;

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

@@ -354,21 +354,29 @@ public static SegmentRing openExisting(FilesFacade ff, String sfDir, long maxByt
             // the next appendOrFsn returns BACKPRESSURE_NO_SPARE, the manager
             // installs a hot spare, the producer rotates. Same fast path as a
             // mid-life ring.
+            //
+            // Ownership transfers to the ring in ONE step, at the very end. Detaching
+            // the active segment from `opened` any earlier -- as an `opened.remove(last)`
+            // before the ring is fully built did -- puts it beyond the reach of the catch
+            // below while that catch can still fire: `new SegmentRing` and the
+            // `sealedSegments.add` loop both allocate, so an ObjList growth IOOBE or a
+            // native OOM lands there with the active segment referenced by nothing. Its
+            // whole-segment mapping (up to sf_max_bytes) and its fd would leak.
             int last = opened.size() - 1;
-            MmapSegment active = opened.get(last);
-            opened.remove(last);
-            SegmentRing ring = new SegmentRing(active, maxBytesPerSegment);
+            SegmentRing ring = new SegmentRing(opened.get(last), maxBytesPerSegment);
             // Older segments become sealed in baseSeq order.
-            for (int i = 0, n = opened.size(); i < n; i++) {
+            for (int i = 0; i < last; i++) {
                 ring.sealedSegments.add(opened.get(i));
             }
+            // Every segment now belongs to `ring`. Release our references LAST, so a
+            // throw anywhere above still finds all of them in `opened`.
+            opened.clear();
             return ring;
         } catch (Throwable t) {
-            // Close every recovered MmapSegment that's still in `opened`.
-            // After the success path, `opened` no longer contains the active
-            // segment (removed above), but the sealed segments transferred to
-            // ring.sealedSegments are still owned by the ring once it's
-            // returned -- so this catch only fires before the return statement.
+            // Close every recovered MmapSegment that's still in `opened`. On the success
+            // path `opened` was cleared only after the ring took ownership of every
+            // segment, so reaching here always means the ring was never returned and
+            // `opened` is still the sole owner -- including of the active segment.
             for (int i = 0, n = opened.size(); i < n; i++) {
                 try {
                     opened.get(i).close();

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

@@ -168,8 +168,33 @@ public static void removeOrphanLogical(FilesFacade ff, String slotDir) {
         if (paths == null) {
             return;
         }
-        ff.remove(paths[1]);
-        ff.remove(paths[2]);
+        // Unlink ONLY while holding the lock. Removing a lock file another party holds
+        // is precisely the attack Files.DIR_MODE_SHARED's javadoc describes: the unlink
+        // does not release that party's flock, but it does free the pathname, so the
+        // next acquirer creates a SECOND inode and locks it successfully -- two owners
+        // of a lock whose only job is mutual exclusion.
+        //
+        // The retiring engine's directory-local lock does not stand in for this. It says
+        // nothing about the LOGICAL lock, which Sender.build() holds across its whole
+        // construct -> connect -> quarantine transition, in a frame ABOVE this one: an
+        // ordinary failed connect closes the engine from inside that scope, reaches here,
+        // and used to unlink the very file build() was holding. Acquiring first turns
+        // that into a no-op instead of silent double-ownership.
+        SlotLock guard;
+        try {
+            guard = acquireAt(ff, slotDir, paths[1], paths[2]);
+        } catch (Throwable t) {
+            // Contended (someone holds it, possibly a caller above us) or unopenable.
+            // Leaving the pair on disk is the safe outcome -- a live holder's lock must
+            // outlive our cleanup. The next fully-drained retirement reclaims it.
+            return;
+        }
+        try {
+            ff.remove(paths[1]);
+            ff.remove(paths[2]);
+        } finally {
+            guard.close();
+        }
     }
 
     private static SlotLock acquireAt(FilesFacade ff, String slotDir, String lockPath, String pidPath) {

core/src/main/java/io/questdb/client/std/Files.java

@@ -80,18 +80,24 @@ public final class Files {
      * create entries in: the store-and-forward root and its logical slot-lock directory.
      * <b>01777 octal -- 0777 plus the sticky bit.</b>
      * <p>
-     * The deployment's umask decides the sharing policy: the usual 022 yields 01755, i.e.
-     * exactly {@link #DIR_MODE_DEFAULT} plus sticky, so the common case is unchanged;
-     * a shared-group deployment (umask 002) gets the 01775 a second uid needs. umask
-     * masks only the rwx bits, never the sticky bit, so the restricted-deletion semantics
-     * survive every umask.
+     * The rwx widening is the load-bearing part and it works: the deployment's umask
+     * decides the sharing policy, so the usual 022 yields 0755 (identical to
+     * {@link #DIR_MODE_DEFAULT}, the common case unchanged) while a shared-group deployment
+     * (umask 002) gets the 0775 a second uid needs to create its own entries here.
      * <p>
-     * The sticky bit is not optional here. Without it, on a multi-tenant host any local
-     * user with write access to the directory can unlink another process's lock file --
-     * which does not release that process's flock, but does free the pathname, so the
-     * next acquirer creates a SECOND inode and locks it successfully. Two owners of a
-     * lock whose entire job is mutual exclusion. Sticky restricts deletion to the file's
-     * owner and closes that.
+     * <b>The sticky bit is best-effort, NOT a guarantee this client relies on.</b> On
+     * Linux {@code mkdir(2)} sets the new directory to {@code (mode & ~umask & 0777)} -- the
+     * {@code & 0777} strips {@code S_ISVTX}, so the bit is silently dropped on the primary
+     * deployment platform (POSIX leaves sticky-on-mkdir unspecified; some platforms honor
+     * it, Linux does not). It is therefore wrong to depend on restricted-deletion semantics
+     * for correctness. Mutual exclusion of the slot lock does NOT rest on it: the safety
+     * comes from never unlinking a lock file this process does not hold
+     * ({@code SlotLock.removeOrphanLogical} acquires the lock before removing it), which
+     * holds regardless of directory permissions. Defending a multi-tenant host against a
+     * hostile local uid unlinking another process's lock would need the sticky bit applied
+     * by an explicit {@code chmod} after {@code mkdir} (and on an already-existing directory
+     * on upgrade, since {@code mkdir} does not touch it) -- there is no {@code chmod} binding
+     * in this class today, so that protection is not provided and must not be assumed.
      * <p>
      * This must be applied to EVERY directory on the path a foreign uid has to create in.
      * Applying it only to {@code .slot-locks} -- while {@code sf_dir} itself stays 0755 --