fix(qwp): make engine close a worker-quiescence barrier before releasing slot resources

SegmentManager.close() gives up after a bounded join, but CursorSendEngine.close()
could not observe the incomplete shutdown: it closed the ring and watermark,
unlinked segment files and released the slot flock while the worker could still
be mid service pass - able to unlink a spare/trim path inside a slot directory
that a replacement engine had already re-acquired (SF data loss after restart).

- serviceRing now claims the entry as in-service under the manager lock and
  skips entries deregistered before the pass starts
- new SegmentManager.awaitRingQuiescence(ring): bounded, interrupt-preserving
  barrier that confirms the worker can never touch the ring/slot again
- CursorSendEngine.close() releases ring/watermark/segment files/slot lock only
  after confirmed quiescence (or a reaped owned worker); otherwise it leaks them
  deliberately, logs, and allows close() to be retried
- a timed-out SegmentManager.close() hands pathScratch ownership to the worker,
  which frees it on exit - no permanent native leak when the worker outlives
  the join
- regression: CursorSendEngineSlotReacquisitionTest (slot retained while worker
  mid-pass + retry completes cleanup; same-slot reacquisition after normal
  close) and an awaitRingQuiescence contract test

Author: bluestreak01

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

@@ -114,6 +114,12 @@ 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.
+    private boolean closeCompleted;
     // Producer-thread-only: timestamp of the last "we're backpressured" log
     // line, used to throttle. Plain long is fine.
     private long lastBackpressureLogNs;
@@ -480,7 +486,7 @@ public long appendOrFsn(long payloadAddr, int payloadLen, long spinDeadlineNanos
 
     @Override
     public synchronized void close() {
-        if (closed) return;
+        if (closed && closeCompleted) return;
         closed = true;
         // Capture drain state BEFORE closing the ring — once the ring is
         // closed, its accessors aren't safe to read. The active segment is
@@ -492,10 +498,14 @@ public synchronized void close() {
         // the server has no dedup state for those messageSequences.
         // Memory mode has no files to unlink.
         // The whole close sequence runs under try/finally so the slot lock
-        // is ALWAYS released, even if manager/ring close or unlink throws —
-        // otherwise a kernel-held flock outlives the engine and the next
-        // sender for the same slot collides on a lock the dead engine
-        // never released.
+        // is released whenever it is safe to do so, even if manager/ring
+        // close or unlink throws — otherwise a kernel-held flock outlives
+        // the engine and the next sender for the same slot collides on a
+        // lock the dead engine never released. The one deliberate exception
+        // is a manager worker that failed to quiesce: releasing the lock
+        // then would let a replacement engine acquire the slot while the
+        // stale worker can still create/unlink segment paths inside it.
+        boolean workerQuiescent = false;
         try {
             // "Fully drained" includes BOTH the obvious case (every published
             // FSN has been acked) AND the never-published case (publishedFsn
@@ -504,9 +514,15 @@ public synchronized void close() {
             // recreates a fresh sf-initial.sfa — would otherwise leave that
             // fresh empty file behind, the next scanner finds it, adopts the
             // slot again, and the cycle repeats forever (M6).
-            boolean fullyDrained = sfDir != null
-                    && (ring.publishedFsn() < 0
-                    || ring.ackedFsn() >= ring.publishedFsn());
+            // Own try/catch so sabotaged/broken ring state cannot skip the
+            // quiescence barrier below or the slot-lock release in finally.
+            boolean fullyDrained = false;
+            try {
+                fullyDrained = sfDir != null
+                        && (ring.publishedFsn() < 0
+                        || ring.ackedFsn() >= ring.publishedFsn());
+            } catch (Throwable ignored) {
+            }
             // Each cleanup step in its own try/catch so a single failure
             // doesn't strand later cleanups — mirrors the constructor's
             // catch block. Without this, a throw from manager.deregister
@@ -517,11 +533,41 @@ public synchronized void close() {
                 manager.deregister(ring);
             } catch (Throwable ignored) {
             }
+            // Quiescence barrier. deregister alone only removes the entry
+            // from the registry — the worker may still be mid service pass
+            // for this ring (creating a spare file, trimming, unlinking).
+            // Releasing the ring, watermark, segment files, or the slot lock
+            // while that pass is in flight lets the stale worker unlink a
+            // path that a replacement engine — which can acquire the slot
+            // the moment the lock is released — is actively writing through:
+            // store-and-forward data loss after restart. Wait for confirmed
+            // quiescence before touching anything the worker can reach.
+            try {
+                workerQuiescent = manager.awaitRingQuiescence(ring);
+            } catch (Throwable ignored) {
+            }
             if (ownsManager) {
                 try {
                     manager.close();
                 } catch (Throwable ignored) {
                 }
+                if (!workerQuiescent) {
+                    // manager.close() joins the worker; a reaped (or
+                    // never-started) worker is an even stronger barrier
+                    // than per-ring quiescence.
+                    try {
+                        workerQuiescent = manager.isWorkerReaped();
+                    } catch (Throwable ignored) {
+                    }
+                }
+            }
+            if (!workerQuiescent) {
+                LOG.error("SF manager worker did not quiesce during engine close; leaking the "
+                        + "ring, watermark and slot lock so a stale worker cannot corrupt a "
+                        + "future engine on slot {}. The kernel releases the slot flock on "
+                        + "process exit; close() may be invoked again to retry cleanup once "
+                        + "the worker has exited.", sfDir == null ? "<memory>" : sfDir);
+                return;
             }
             try {
                 ring.close();
@@ -551,8 +597,13 @@ public synchronized void close() {
                 } catch (Throwable ignored) {
                 }
             }
+            closeCompleted = true;
         } finally {
-            if (slotLock != null) {
+            // Gate on quiescence: releasing the flock while a stale worker
+            // can still touch this slot directory hands the slot to a new
+            // engine that the worker may then corrupt. Leaking the fd is
+            // the safe failure mode — the kernel drops it on process exit.
+            if (workerQuiescent && slotLock != null) {
                 try {
                     slotLock.close();
                 } catch (Throwable ignored) {

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

@@ -100,15 +100,34 @@ public final class SegmentManager implements QuietCloseable {
     // registered flag check inside the trim block closes for watermark writes
     // and totalBytes accounting.
     private volatile Runnable beforeTrimSyncHook;
+    // Entry currently being serviced by the worker thread, or null when the
+    // worker is between service passes (or not running). Guarded by
+    // {@link #lock}; cleared with lock.notifyAll() so awaitRingQuiescence can
+    // block until an in-flight pass for a just-deregistered ring finishes.
+    private RingEntry inService;
     private long lastDiskFullLogNs;
     private volatile boolean running;
+    // pathScratch free-exactly-once coordination between a timed-out close()
+    // and the worker's exit path. All three are guarded by {@link #lock}.
+    // When close() gives up on the join while the worker loop has not yet
+    // exited, it hands scratch ownership to the worker
+    // (scratchHandedToWorker=true) and the worker frees the buffer in its
+    // exit block; in every other case close() frees it. Without the handoff,
+    // a worker that outlives the bounded join leaks the native scratch
+    // buffer forever, because nobody retries manager cleanup after close()
+    // returns.
+    private boolean scratchFreed;
+    private boolean scratchHandedToWorker;
+    private boolean workerLoopExited;
     // Total bytes currently allocated across every segment owned by every
     // registered ring (active + sealed + hot-spare). Mutated by the manager
     // thread on provision/trim and by register/deregister callers under
     // {@link #lock}; the lock covers both paths so the counter stays
     // consistent across registration boundaries.
     private long totalBytes;
-    private long workerJoinTimeoutMillis = WORKER_JOIN_TIMEOUT_MILLIS;
+    // volatile: read by awaitRingQuiescence() from arbitrary caller threads
+    // while the @TestOnly setter may run on another.
+    private volatile long workerJoinTimeoutMillis = WORKER_JOIN_TIMEOUT_MILLIS;
     // volatile because wakeWorker() reads workerThread without holding the
     // monitor; the synchronized start()/close() pair handles the
     // start-vs-close ordering.
@@ -194,25 +213,114 @@ public synchronized void close() {
                 }
             }
             if (t.isAlive()) {
-                LOG.warn("SegmentManager worker did not stop before close wait completed; "
-                        + "leaving worker-owned resources allocated");
-                return;
+                synchronized (lock) {
+                    if (!workerLoopExited) {
+                        // Hand pathScratch ownership to the worker: its exit
+                        // block frees the buffer under the same lock, so the
+                        // native allocation is reclaimed even though this
+                        // close() could not confirm termination. workerThread
+                        // stays set so isWorkerReaped() reports the incomplete
+                        // shutdown and a later close() can retry the join.
+                        scratchHandedToWorker = true;
+                        LOG.warn("SegmentManager worker did not stop before close wait completed; "
+                                + "worker frees its native scratch buffer on exit");
+                        return;
+                    }
+                }
+                // The worker loop has already run its exit block; the thread
+                // is at most a few instructions from terminating and can no
+                // longer touch manager state. Fall through and reap it.
             }
             workerThread = null;
         }
         // Free the rotation-path native scratch buffer only after worker
-        // termination has been observed. The worker is the only thread that
-        // touches the buffer, but close() uses a bounded join; if the worker is
-        // still alive, leaking this one scratch allocation is safer than freeing
-        // native memory it may still read or write.
-        pathScratch.close();
+        // termination (or worker-loop exit) has been observed. The worker is
+        // the only thread that touches the buffer; the scratchFreed flag
+        // (shared with the worker's exit block) makes the free exactly-once
+        // no matter which side runs last.
+        synchronized (lock) {
+            if (!scratchFreed) {
+                scratchFreed = true;
+                pathScratch.close();
+            }
+        }
+    }
+
+    /**
+     * Quiescence barrier for {@link #deregister(SegmentRing)}. Blocks until
+     * the worker thread is provably no longer executing a service pass for
+     * {@code ring}, or the worker-join timeout elapses. After this returns
+     * {@code true}, the worker will never again touch the ring, its
+     * watermark, or path names under its slot directory: deregister has
+     * removed the entry from the registry, a stale snapshot entry that has
+     * not started its pass is skipped by the registration check at the top
+     * of {@link #serviceRing(RingEntry)}, and this method has observed the
+     * end of any in-flight pass. Only then may the caller release dependent
+     * resources (ring, watermark, segment files, slot lock).
+     * <p>
+     * Returns {@code true} immediately when no worker is running or when
+     * called from the worker thread itself (test hooks inject deregister
+     * calls there; waiting would self-deadlock). Returns {@code false} when
+     * the in-flight pass did not finish within the timeout — the caller must
+     * treat the worker as still live and leak rather than release.
+     * <p>
+     * A pending caller interrupt is preserved but does not abort the wait,
+     * mirroring {@link #close()}.
+     */
+    public boolean awaitRingQuiescence(SegmentRing ring) {
+        Thread t = workerThread;
+        if (t == null || t == Thread.currentThread()) {
+            return true;
+        }
+        long deadlineNanos = System.nanoTime() + workerJoinTimeoutMillis * 1_000_000L;
+        boolean interrupted = Thread.interrupted();
+        try {
+            synchronized (lock) {
+                while (inService != null && inService.ring == ring) {
+                    long remainingNanos = deadlineNanos - System.nanoTime();
+                    if (remainingNanos <= 0) {
+                        return false;
+                    }
+                    try {
+                        // Round up so a sub-millisecond remainder still waits
+                        // instead of spinning through wait(0) == wait-forever.
+                        lock.wait(Math.max(1L, remainingNanos / 1_000_000L));
+                    } catch (InterruptedException ignored) {
+                        interrupted = true;
+                    }
+                }
+            }
+            return true;
+        } finally {
+            if (interrupted) {
+                Thread.currentThread().interrupt();
+            }
+        }
+    }
+
+    /**
+     * True when no manager worker thread can be running: either
+     * {@link #start()} was never called, or a {@link #close()} confirmed
+     * worker termination and reaped the thread. Owners use this as a
+     * stronger fallback barrier when {@link #awaitRingQuiescence(SegmentRing)}
+     * times out but a subsequent {@code close()} join succeeded.
+     */
+    public synchronized boolean isWorkerReaped() {
+        return workerThread == null;
     }
 
     /**
      * Stop tracking {@code ring}. Pending spares for the ring are NOT
      * created after this returns, but already-installed spares stay with
      * the ring (the ring closes them on its own {@link SegmentRing#close}).
      * Idempotent; safe to call from any thread.
+     * <p>
+     * Non-blocking: a worker service pass already in flight for this ring
+     * may still be running when this returns. Callers about to release
+     * resources the worker can reach (the ring itself, its watermark, its
+     * segment files, or the slot lock guarding its directory) MUST follow
+     * up with {@link #awaitRingQuiescence(SegmentRing)} and only release on
+     * a {@code true} result.
      */
     public void deregister(SegmentRing ring) {
         synchronized (lock) {
@@ -412,6 +520,32 @@ private String nextSparePath(String dir) {
     }
 
     private void serviceRing(RingEntry e) {
+        // Claim the entry as in-service so deregister-side quiescence
+        // barriers (awaitRingQuiescence) can wait for this pass to finish.
+        // A stale snapshot entry deregistered before the pass starts is
+        // skipped entirely: the deregistering thread may already be
+        // releasing the ring / watermark / slot resources, so the worker
+        // must not touch them at all. The registered check and the
+        // in-service claim are atomic under `lock` — deregister flips
+        // `registered` under the same lock, so it either prevents this
+        // pass or the barrier observes it via `inService`.
+        synchronized (lock) {
+            if (!e.registered) {
+                return;
+            }
+            inService = e;
+        }
+        try {
+            serviceRing0(e);
+        } finally {
+            synchronized (lock) {
+                inService = null;
+                lock.notifyAll();
+            }
+        }
+    }
+
+    private void serviceRing0(RingEntry e) {
         // 1. Provision a hot spare if the ring needs one AND we have headroom
         //    under the disk-total cap. Cap check is per-tick; if we're capped
         //    here, the ring stays in BACKPRESSURE_NO_SPARE until trim (step 2)
@@ -571,26 +705,40 @@ private void serviceRing(RingEntry e) {
     }
 
     private void workerLoop() {
-        while (running) {
-            // Snapshot the rings under the lock so we don't hold it through the
-            // (potentially slow) syscalls during creation/unlink. ringSnapshot
-            // is a thread-confined field — no per-tick allocation.
-            ringSnapshot.clear();
-            synchronized (lock) {
-                for (int i = 0, n = rings.size(); i < n; i++) {
-                    ringSnapshot.add(rings.getQuick(i));
+        try {
+            while (running) {
+                // Snapshot the rings under the lock so we don't hold it through the
+                // (potentially slow) syscalls during creation/unlink. ringSnapshot
+                // is a thread-confined field — no per-tick allocation.
+                ringSnapshot.clear();
+                synchronized (lock) {
+                    for (int i = 0, n = rings.size(); i < n; i++) {
+                        ringSnapshot.add(rings.getQuick(i));
+                    }
                 }
-            }
-            for (int i = 0, n = ringSnapshot.size(); i < n; i++) {
+                for (int i = 0, n = ringSnapshot.size(); i < n; i++) {
+                    if (!running) break;
+                    serviceRing(ringSnapshot.getQuick(i));
+                }
+                // Drop strong refs so a deregistered ring becomes collectable
+                // before the next tick (otherwise the snapshot pins it for up
+                // to pollNanos after deregister).
+                ringSnapshot.clear();
                 if (!running) break;
-                serviceRing(ringSnapshot.getQuick(i));
+                LockSupport.parkNanos(pollNanos);
+            }
+        } finally {
+            // If a timed-out close() abandoned the reap, it handed
+            // pathScratch ownership to this thread (see close()). Freeing it
+            // here reclaims the native buffer even when the worker outlives
+            // every close() attempt — nobody else retries manager cleanup.
+            synchronized (lock) {
+                workerLoopExited = true;
+                if (scratchHandedToWorker && !scratchFreed) {
+                    scratchFreed = true;
+                    pathScratch.close();
+                }
             }
-            // Drop strong refs so a deregistered ring becomes collectable
-            // before the next tick (otherwise the snapshot pins it for up
-            // to pollNanos after deregister).
-            ringSnapshot.clear();
-            if (!running) break;
-            LockSupport.parkNanos(pollNanos);
         }
     }