fix(qwp): preserve slot ownership after incomplete close

Author: bluestreak01

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

@@ -278,10 +278,10 @@ public class QwpWebSocketSender implements Sender {
     private boolean ownsCursorEngine;
     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
-    // still running -- then cursorEngine.close() never ran and the flock is
-    // still held, so the owning pool MUST keep the slot reserved rather than
-    // hand the still-locked dir to the next borrow ("sf slot already in use").
+    // teardown path). Stays false if an I/O or manager worker did not stop and
+    // cursorEngine retained the flock, so the owning pool MUST keep the slot
+    // reserved rather than hand the still-locked dir to the next borrow
+    // ("sf slot already in use").
     private boolean slotLockReleased;
     private int pendingRowCount;
     private SenderProgressDispatcher progressDispatcher;
@@ -1038,7 +1038,10 @@ public QwpWebSocketSender charColumn(CharSequence columnName, char value) {
      *       replaying frames get a 2.5s grace window plus a 0.5s stop window
      *       — worst case ~3s when a drainer sits in a blocking native
      *       connect (15s background deadline) and must be abandoned to exit
-     *       on its own.</li>
+     *       on its own;</li>
+     *   <li>SF manager stop: normally immediate, bounded by 5s when its worker
+     *       is stuck in a filesystem operation. On timeout the slot remains
+     *       locked rather than being exposed to a stale worker.</li>
      * </ul>
      */
     @Override
@@ -1189,15 +1192,23 @@ public void close() {
                 // the failed close() and now — then closing here is safe.
                 if (ownsCursorEngine && cursorEngine != null && cursorSendLoop != null
                         && !cursorSendLoop.delegateEngineClose()) {
+                    CursorSendEngine engine = cursorEngine;
                     try {
-                        cursorEngine.close();
+                        engine.close();
                     } catch (Throwable t) {
                         LOG.error("Error closing owned CursorSendEngine: {}", String.valueOf(t));
                         terminalError = captureCloseError(terminalError, t);
                     }
-                    cursorEngine = null;
-                    ownsCursorEngine = false;
-                    slotLockReleased = true;
+                    if (engine.isCloseCompleted()) {
+                        cursorEngine = null;
+                        ownsCursorEngine = false;
+                        slotLockReleased = true;
+                    } else {
+                        // Preserve the engine and report the retained flock.
+                        // Sender.close() is idempotent, so this is a deliberate
+                        // safe leak until process exit rather than a retry path.
+                        slotLockReleased = false;
+                    }
                 }
                 rethrowTerminal(terminalError);
                 return;
@@ -1231,19 +1242,27 @@ public void close() {
             }
 
             if (ownsCursorEngine && cursorEngine != null) {
+                CursorSendEngine engine = cursorEngine;
                 try {
-                    cursorEngine.close();
+                    engine.close();
                 } catch (Throwable t) {
                     LOG.error("Error closing owned CursorSendEngine: {}", String.valueOf(t));
                     terminalError = captureCloseError(terminalError, t);
                 }
-                cursorEngine = null;
-                ownsCursorEngine = false;
+                if (engine.isCloseCompleted()) {
+                    cursorEngine = null;
+                    ownsCursorEngine = false;
+                    slotLockReleased = true;
+                } else {
+                    // The manager worker did not quiesce. Preserve ownership
+                    // and report the retained flock so pools retire this slot.
+                    // Repeated Sender.close() calls remain no-ops by contract.
+                    slotLockReleased = false;
+                }
+            } else {
+                // This sender owns no cursor engine holding an SF flock.
+                slotLockReleased = true;
             }
-            // Past the ioThreadStopped guard => cursorEngine.close() ran and
-            // released the SF flock in its finally (or this sender owned no
-            // engine holding one). Signal the pool it may reuse the slot.
-            slotLockReleased = true;
 
             // Shutdown order: dispatcher last, after the I/O loop has stopped
             // producing into it. close() drains pending entries with a short
@@ -1288,9 +1307,9 @@ public void close() {
 
     /**
      * True once {@link #close()} has released the store-and-forward slot
-     * flock. False means close() leaked the still-running I/O thread (and its
-     * resources), so the flock is still held; the owning pool must keep the
-     * slot index reserved instead of reusing the still-locked slot dir.
+     * flock. False means an I/O or manager worker did not stop and close()
+     * retained the lock and worker-reachable resources; the owning pool must
+     * keep the slot index reserved instead of reusing the still-locked dir.
      */
     public boolean isSlotLockReleased() {
         return slotLockReleased;

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

@@ -540,25 +540,26 @@ public synchronized void close() {
             // 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) {
-            }
+            // store-and-forward data loss after restart.
             if (ownsManager) {
+                // Stopping and reaping a private manager is a stronger barrier
+                // than waiting for this ring alone. Do it directly so a stuck
+                // worker consumes at most one workerJoinTimeoutMillis budget,
+                // rather than one here and a second one in manager.close().
                 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) {
-                    }
+                try {
+                    workerQuiescent = manager.isWorkerReaped();
+                } catch (Throwable ignored) {
+                }
+            } else {
+                // A shared manager must keep serving its other rings, so wait
+                // only for the deregistered ring's current pass to finish.
+                try {
+                    workerQuiescent = manager.awaitRingQuiescence(ring);
+                } catch (Throwable ignored) {
                 }
             }
             if (!workerQuiescent) {
@@ -613,6 +614,16 @@ public synchronized void close() {
         }
     }
 
+    /**
+     * 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. Owners must not report or reuse the slot in that case.
+     */
+    public synchronized boolean isCloseCompleted() {
+        return closeCompleted;
+    }
+
     /**
      * Pass-through to {@link SegmentRing#findSegmentContaining(long)}.
      */

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

@@ -93,6 +93,10 @@ public final class SegmentManager implements QuietCloseable {
     // but before ownership/accounting commit. Callers may inject a deregister
     // or hold this stale worker snapshot while caller-side cleanup runs.
     private volatile Runnable beforeInstallSyncHook;
+    // Test seam: records entry into the per-ring quiescence wait. Null in
+    // production; owned-engine close tests use it to prove they take only the
+    // stronger whole-manager join path, not two sequential timeout budgets.
+    private volatile Runnable beforeRingQuiescenceAwaitHook;
     // Test seam: runs on the worker thread just before the trim block's
     // synchronized(lock) entry. Null in production; only
     // SegmentManagerTrimDeregisterRaceTest installs it, to deterministically
@@ -268,6 +272,10 @@ public synchronized void close() {
      * mirroring {@link #close()}.
      */
     public boolean awaitRingQuiescence(SegmentRing ring) {
+        Runnable hook = beforeRingQuiescenceAwaitHook;
+        if (hook != null) {
+            hook.run();
+        }
         Thread t = workerThread;
         if (t == null || t == Thread.currentThread()) {
             return true;
@@ -410,6 +418,11 @@ public void setBeforeInstallSyncHook(Runnable hook) {
         this.beforeInstallSyncHook = hook;
     }
 
+    @TestOnly
+    public void setBeforeRingQuiescenceAwaitHook(Runnable hook) {
+        this.beforeRingQuiescenceAwaitHook = hook;
+    }
+
     @TestOnly
     public void setBeforeTrimSyncHook(Runnable hook) {
         this.beforeTrimSyncHook = hook;

core/src/main/java/io/questdb/client/impl/SenderPool.java

@@ -171,7 +171,7 @@ public final class SenderPool implements AutoCloseable {
     // down on another thread. Guarded by lock.
     private int pendingLeaseTeardowns;
     // Slots whose delegate close() returned with the SF flock still held
-    // (the I/O thread refused to stop). Permanently consumed: the index is
+    // because an I/O or manager worker did not stop. Permanently consumed:
     // never freed and never reused, so no borrow ever hands out a still-
     // locked slot dir. Counted in the cap check so the lost capacity is
     // accounted for. Guarded by lock; only ever ticks for SF slots.
@@ -523,15 +523,15 @@ private boolean recoverOneSlotStep(long stepBudgetMillis) {
                 // leakedSlots (retire) or the freed index carries the cap math.
                 recoveringSlots--;
                 if (flockHeld[0]) {
-                    // close() bailed early with the I/O thread still running and
-                    // the flock still held. Retire the slot permanently (mirror
+                    // close() retained the flock because an I/O or manager
+                    // worker did not stop. Retire the slot permanently (mirror
                     // discardBroken/reapIdle): keep slotInUse[i] set and count it
                     // in leakedSlots so the borrow() cap math accounts for the
                     // lost capacity and no later borrow ever reuses the
                     // still-locked dir.
                     leakedSlots++;
                     LOG.warn("startup SF recovery: slot {} retired permanently: delegate close() returned with "
-                                    + "the flock still held (I/O thread refused to stop); pool capacity reduced by 1, "
+                                    + "the flock still held (I/O or manager worker did not stop); pool capacity reduced by 1, "
                                     + "now {} of {} usable [leakedSlots={}]",
                             i, maxSize - leakedSlots, maxSize, leakedSlots);
                 } else {
@@ -580,12 +580,13 @@ private boolean recoverOneSlotStep(long stepBudgetMillis) {
             if (flockHeld[0]) {
                 // Out of the pool's [0, maxSize) capacity range: there is no
                 // slotInUse entry to retire and no future borrow targets this
-                // dir, so a still-held flock only leaks this recoverer's I/O
-                // thread (a best-effort teardown loss, logged). Crucially we do
+                // dir, so a still-held flock only leaks this recoverer's
+                // worker-reachable resources (a best-effort teardown loss,
+                // logged). Crucially we do
                 // NOT touch leakedSlots -- that would wrongly shrink the
                 // in-range pool capacity.
                 LOG.warn("startup SF recovery: out-of-range slot {} closed with the flock still held "
-                                + "(I/O thread refused to stop); its data is durable on disk for a later attempt",
+                                + "(I/O or manager worker did not stop); its data is durable on disk for a later attempt",
                         slotPath);
             }
             if (stopScan) {
@@ -609,7 +610,7 @@ private boolean recoverOneSlotStep(long stepBudgetMillis) {
      *
      * @param flockHeld single-element out-param set to {@code true} iff a
      *                  recoverer was built and its {@code close()} returned with
-     *                  the flock still held (the I/O thread refused to stop)
+     *                  the flock still held because a worker did not stop
      * @return {@code true} if a build/drain failure occurred that will very
      * likely repeat for every remaining slot, so the caller should stop scanning
      */
@@ -618,8 +619,8 @@ private boolean drainCandidateSlotForRecovery(int slotIndex, String slotPath,
         flockHeld[0] = false;
         // Hoisted so the flock check after the try can consult it:
         // createRecoverer() takes the slot flock on <base>-slotIndex, and
-        // delegate().close() can early-return with the I/O thread still running
-        // (flock still held).
+        // delegate().close() can retain it when an I/O or manager worker does
+        // not stop.
         SenderSlot recoverer = null;
         boolean stopScan = false;
         try {
@@ -1067,7 +1068,7 @@ public void reapIdle() {
             }
             // Return reserved SF slot indices to the free set -- but only for
             // slots whose delegate confirmed the flock was released. A slot
-            // left locked (I/O thread refused to stop) is retired permanently.
+            // left locked because a worker did not stop is retired permanently.
             if (storeAndForward) {
                 lock.lock();
                 try {
@@ -1107,8 +1108,8 @@ public int totalSize() {
 
     /**
      * Snapshot of the number of SF slots permanently retired because a
-     * delegate {@code close()} returned with the slot flock still held (the
-     * I/O thread refused to stop). Each leaked slot permanently lowers the
+     * delegate {@code close()} returned with the slot flock still held after
+     * an I/O or manager worker did not stop. Each leaked slot permanently lowers the
      * pool's effective capacity ({@code maxSize - leakedSlotCount()}). A
      * non-zero, growing value explains a pool that has started timing out
      * every {@code borrow()}. For metrics and tests.
@@ -1271,7 +1272,7 @@ private void freeSlotIndex(int idx) {
      * non-QWP delegate never holds an SF flock, so it is always treated as
      * released. A {@link QwpWebSocketSender} reports it via
      * {@link QwpWebSocketSender#isSlotLockReleased()} -- false means close()
-     * bailed early with the I/O thread still running and the flock still held.
+     * retained the flock because an I/O or manager worker did not stop.
      */
     private static boolean flockReleased(SenderSlot s) {
         Sender d = s.delegate();
@@ -1281,8 +1282,8 @@ private static boolean flockReleased(SenderSlot s) {
     /**
      * Reclaims one SF slot after its delegate's {@code close()} has been
      * attempted. When the flock was released the index returns to the free
-     * set; when {@code close()} returned with the flock still held (the I/O
-     * thread refused to stop) the slot is retired permanently --
+     * set; when {@code close()} returned with the flock still held because an
+     * I/O or manager worker did not stop, the slot is retired permanently --
      * {@code leakedSlots++} and {@code slotInUse[idx]} stays set -- so the cap
      * math accounts for the lost capacity and no later borrow ever reuses the
      * still-locked dir. Either way {@code closingSlots} is decremented.
@@ -1304,7 +1305,7 @@ private boolean reclaimSlot(SenderSlot s, String context) {
         }
         leakedSlots++;
         LOG.warn("SF slot {} retired permanently{}: delegate close() returned with the flock still held " +
-                        "(I/O thread refused to stop); pool capacity reduced by 1, now {} of {} usable [leakedSlots={}]",
+                        "(I/O or manager worker did not stop); pool capacity reduced by 1, now {} of {} usable [leakedSlots={}]",
                 s.slotIndex(), context, maxSize - leakedSlots, maxSize, leakedSlots);
         return false;
     }