fix(qwp): publish close completion only after confirmed slot flock release

finishClose() wrote closeCompleted=true before slotLock.close(), so a
pool thread could observe completion (isSlotLockReleased ->
reprobeRetiredSlots), free the slot index, and admit a replacement
sender whose SlotLock.acquire collided with the still-open flock fd --
a spurious "sf slot already in use" naming the process's own pid.
SlotLock.close() also discarded the Files.close() result, reporting an
unconfirmed release as completion.

Reorder the terminal cleanup: release the flock first via the new
SlotLock.release() (checks the close rc, retains the fd on failure so
the lock state is never misreported), and publish closeCompleted only
on a confirmed release. An unconfirmed release keeps closeCompleted
false, degrading into the existing retired/leaked-slot accounting with
the kernel's process-exit backstop.

Add a @testonly hook between cleanup and release so the otherwise
microsecond-wide window is deterministically testable; the new test
parks the closer inside it and asserts completion is not observable
while the flock is provably still held, then latches once released.

Author: bluestreak01

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

@@ -28,6 +28,7 @@
 import io.questdb.client.std.Files;
 import io.questdb.client.std.ObjList;
 import io.questdb.client.std.QuietCloseable;
+import org.jetbrains.annotations.TestOnly;
 
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.locks.LockSupport;
@@ -115,15 +116,28 @@ public final class CursorSendEngine implements QuietCloseable {
     // thread, JVM shutdown hooks, test cleanup). volatile + synchronized
     // close() makes the check-and-set atomic and gives readers a fence.
     private volatile boolean closed;
-    // True once close() has run its full cleanup sequence. Stays false when
-    // a close attempt could not confirm manager-worker quiescence and had to
-    // leak the ring/watermark/slot lock — in that case a later close() call
-    // retries the cleanup (the worker may have exited by then). Guarded by
-    // the synchronized close() method; never read elsewhere.
+    // True once close() has run its full cleanup sequence INCLUDING a
+    // CONFIRMED slot-flock release — finishClose() publishes this strictly
+    // after SlotLock.release() reports success, never before. Pool threads
+    // treat the flip as "the slot dir is reusable" and free the slot index
+    // the moment they observe it (QwpWebSocketSender.isSlotLockReleased ->
+    // SenderPool.reprobeRetiredSlots), so publishing before the release
+    // would let a replacement engine's SlotLock.acquire collide with the
+    // still-open fd. Stays false when a close attempt could not confirm
+    // manager-worker quiescence (or the flock release itself failed) and had
+    // to leak the ring/watermark/slot lock — in that case a later close()
+    // call retries the cleanup (the worker may have exited by then).
     // volatile: latched by finishClose(), but read lock-free by
     // isCloseCompleted() from pool threads re-probing a retired slot (see the
     // getter for why it must not synchronize).
     private volatile boolean closeCompleted;
+    // Test-only hook run by finishClose() between the terminal cleanup and
+    // the flock release. Lets a test park the releasing thread inside the
+    // cleanup/release window and assert that closeCompleted stays false —
+    // i.e. that completion is never observable while the flock is still
+    // held. volatile: finishClose may run on the manager worker's exit
+    // thread while the hook is installed from a test thread.
+    private volatile Runnable beforeFlockReleaseHook;
     // Exactly-once claim on the terminal cleanup (finishClose). Contended by
     // close() and a worker-exit handoff (completeDeferredClose); whoever wins
     // the CAS runs the cleanup, the loser never touches ring/watermark/flock.
@@ -586,9 +600,17 @@ public synchronized void close() {
             // cleanup runs exactly once and the worker never blocks on the
             // engine monitor a retried close() holds while joining it.
             boolean handedOff = false;
+            boolean registrationFailed = false;
             try {
                 handedOff = manager.deferUntilWorkerExit(() -> completeDeferredClose(fullyDrained));
             } catch (Throwable ignored) {
+                // Allocation failure (OOM building the cleanup lambda or
+                // growing the manager's exitCleanups list). Unlike the exact
+                // false return below, a throw carries NO worker-liveness
+                // information — the two must never be conflated, or the
+                // inline cleanup below would release worker-reachable
+                // resources under a possibly-live worker.
+                registrationFailed = true;
             }
             if (handedOff) {
                 LOG.error("SF manager worker did not quiesce during engine close; ring, watermark "
@@ -598,6 +620,23 @@ public synchronized void close() {
                         + "the stale worker on slot {}", sfDir == null ? "<memory>" : sfDir);
                 return;
             }
+            if (registrationFailed) {
+                // The handoff never registered and the worker was never
+                // observed — it must be presumed live and mid service pass.
+                // Retain every worker-reachable resource (ring, watermark,
+                // segment files, slot flock) and leave terminalCleanupClaimed
+                // unclaimed and closeCompleted false, exactly like the
+                // shared-manager leak branch: manager.close() above already
+                // stopped the worker loop, so a retried close() converges via
+                // isWorkerReaped() once the in-flight pass ends. The kernel
+                // releases the slot flock on process exit regardless.
+                LOG.error("SF worker-exit handoff registration failed during engine close; "
+                        + "leaking the ring, watermark and slot lock so a possibly-live "
+                        + "worker cannot corrupt a future engine on slot {}. close() may be "
+                        + "invoked again to retry cleanup once the worker has exited.",
+                        sfDir == null ? "<memory>" : sfDir);
+                return;
+            }
             // Handoff rejected: the worker loop exited between the failed
             // bounded join and the registration attempt (both sides share the
             // manager's lock, so this observation is exact). A worker past
@@ -629,8 +668,13 @@ public synchronized void close() {
 
     /**
      * Terminal cleanup: closes the ring and watermark, unlinks drained
-     * segment files, releases the slot flock, and latches
-     * {@link #closeCompleted}. The caller must hold the engine monitor and
+     * segment files, releases the slot flock, and — only once the release
+     * is <b>confirmed</b> — latches {@link #closeCompleted}. Publish order
+     * is load-bearing: pools free the slot index the instant they observe
+     * {@code closeCompleted} (via {@code isSlotLockReleased()}), so it must
+     * never be visible while the flock fd is still open, or a replacement
+     * engine races the release and fails acquisition on a live slot.
+     * The caller must hold the engine monitor and
      * must have established that the manager worker can no longer touch the
      * slot directory (reaped, provably exited, or running this on its own
      * exit path) AND have won the {@link #terminalCleanupClaimed} CAS — the
@@ -669,21 +713,64 @@ private void finishClose(boolean fullyDrained) {
                 } catch (Throwable ignored) {
                 }
             }
-            closeCompleted = true;
         } finally {
             // Reaching finishClose at all requires established quiescence, so
             // releasing the flock is safe even if a step above threw. Leaking
             // it would strand the slot until process exit for no reason.
+            //
+            // ORDER MATTERS: release the flock FIRST, verify it, and only
+            // then publish closeCompleted. Pools read isCloseCompleted() as
+            // "the slot dir is reusable" and free the slot index the moment
+            // it flips; publishing before Files.close(fd) completes would
+            // open a window where a replacement engine's SlotLock.acquire
+            // collides with the still-open fd and fails with a spurious
+            // "slot already in use".
+            Runnable hook = beforeFlockReleaseHook;
+            if (hook != null) {
+                try {
+                    hook.run();
+                } catch (Throwable ignored) {
+                    // test-only; must never block the release
+                }
+            }
+            boolean released;
             if (slotLock != null) {
                 try {
-                    slotLock.close();
+                    released = slotLock.release();
                 } catch (Throwable ignored) {
-                    // best-effort; flock is also released by kernel on process exit
+                    released = false;
                 }
+            } else {
+                released = true;
+            }
+            if (released) {
+                closeCompleted = true;
+            } else {
+                // Unconfirmed release: the flock may still be held, so keep
+                // closeCompleted false — the owning pool keeps the slot
+                // retired (leakedSlots accounting) instead of reusing a
+                // possibly-locked dir. The kernel releases the flock on
+                // process exit regardless.
+                LOG.error("SF slot flock release failed during engine close; keeping "
+                        + "closeCompleted=false so the pool cannot reuse a possibly "
+                        + "still-locked slot dir; the kernel releases the flock on "
+                        + "process exit [slot={}]", sfDir == null ? "<memory>" : sfDir);
             }
         }
     }
 
+    /**
+     * Installs a hook that {@link #finishClose} runs between the terminal
+     * cleanup and the slot-flock release. Test-only: makes the otherwise
+     * microsecond-wide cleanup/release window deterministic so tests can
+     * assert {@link #isCloseCompleted()} stays false until the release is
+     * confirmed.
+     */
+    @TestOnly
+    public void setBeforeFlockReleaseHook(Runnable hook) {
+        this.beforeFlockReleaseHook = hook;
+    }
+
     /**
      * Runs on the manager worker's exit path when {@link #close()} handed
      * cleanup ownership to the worker (see {@code deferUntilWorkerExit}).
@@ -705,10 +792,13 @@ private void completeDeferredClose(boolean fullyDrained) {
     }
 
     /**
-     * Whether {@link #close()} completed all cleanup, including releasing the
-     * SF slot lock. A false value after close means manager-worker quiescence
-     * could not be confirmed and the worker-reachable resources were retained
-     * deliberately — either handed to the worker's exit path (owned manager),
+     * Whether {@link #close()} completed all cleanup, including a
+     * <b>confirmed</b> release of the SF slot lock — the flip is published
+     * strictly after the flock fd is closed, so observing {@code true}
+     * guarantees the slot dir is acquirable by a replacement engine. A false
+     * value after close means manager-worker quiescence could not be
+     * confirmed (or the flock release itself failed) and the
+     * worker-reachable resources were retained deliberately — either handed to the worker's exit path (owned manager),
      * which flips this to true the moment the worker's in-flight pass
      * finishes, or leaked until a retried close() (shared manager). Owners
      * must not reuse the slot while this is false; pools may re-probe it to

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

@@ -116,15 +116,43 @@ public String slotDir() {
         return slotDir;
     }
 
-    @Override
-    public void close() {
-        // Closing the fd releases the lock. We do NOT remove the .lock
-        // file or the .lock.pid sidecar — a stale PID is harmless (next
-        // acquirer overwrites .lock.pid on success).
-        if (fd >= 0) {
-            Files.close(fd);
+    /**
+     * Releases the flock by closing the lock fd and reports whether the
+     * release was <b>confirmed</b>. Closing the fd releases the lock. We do
+     * NOT remove the {@code .lock} file or the {@code .lock.pid} sidecar —
+     * a stale PID is harmless (next acquirer overwrites {@code .lock.pid}
+     * on success).
+     * <p>
+     * On close failure the fd is <b>retained</b>: the flock may still be
+     * held by this process, so forgetting the fd would misreport the lock
+     * state and forfeit any chance of a later retry. Idempotent — once
+     * released, subsequent calls return {@code true}.
+     * <p>
+     * Owners that gate a "slot dir is reusable" signal on the release
+     * (e.g. {@code CursorSendEngine.finishClose} publishing
+     * {@code closeCompleted}) must call this and check the result rather
+     * than {@link #close()}, which is best-effort by contract.
+     *
+     * @return {@code true} if the fd was closed (or was already released),
+     *         {@code false} if the OS reported a close failure and the
+     *         flock may still be held
+     */
+    public boolean release() {
+        if (fd < 0) {
+            return true;
+        }
+        if (Files.close(fd) == 0) {
             fd = -1;
+            return true;
         }
+        return false;
+    }
+
+    @Override
+    public void close() {
+        // QuietCloseable contract: best-effort, no signal. Callers that
+        // must confirm the release use release() and check the result.
+        release();
     }
 
     private static String readHolder(String pidPath) {