Harden QWP recovery and shutdown

Cursor engine closure now retains slot ownership until manager
quiescence and transfers stalled cleanup to a dedicated daemon.

Recovery validates persisted segments through bounded positional reads
before mmap, preserves facade ownership, and retains corrupt data for
diagnosis.

Deterministic tests cover lifecycle races, compatibility, recovery
faults, and slot reuse.

Author: bluestreak01

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

@@ -219,6 +219,9 @@ public class QwpWebSocketSender implements Sender {
     // into the engine's mmap'd ring; the cursorSendLoop is the I/O thread
     // that walks the ring and sends frames.
     private CursorSendEngine cursorEngine;
+    // True after a failed I/O-thread stop handed engine ownership to that
+    // thread's exit path. Repeated sender close calls must not race that owner.
+    private boolean cursorEngineCloseDelegated;
     private CursorWebSocketSendLoop cursorSendLoop;
     private boolean deferCommit;
     // User-supplied observer for background orphan-slot drainer events.
@@ -1043,8 +1046,20 @@ public QwpWebSocketSender charColumn(CharSequence columnName, char value) {
      */
     @Override
     public void close() {
-        if (!closed) {
-            closed = true;
+        if (closed) {
+            // A prior close may have timed out waiting for manager quiescence.
+            // Retry only when this thread still owns cleanup; delegated engine
+            // closes belong exclusively to the I/O thread's exit path.
+            if (!cursorEngineCloseDelegated) {
+                try {
+                    closeOwnedCursorEngine();
+                } catch (Throwable t) {
+                    LOG.error("Error retrying owned CursorSendEngine close: {}", String.valueOf(t));
+                }
+            }
+            return;
+        }
+        closed = true;
             boolean ioThreadStopped = true;
             // Captures the first error from the flush/drain path AND any
             // secondary errors from cleanup steps (added via addSuppressed).
@@ -1187,17 +1202,17 @@ public void close() {
                 // close actually runs, so the pool must not reuse the slot
                 // meanwhile. A false return means the thread exited between
                 // the failed close() and now — then closing here is safe.
-                if (ownsCursorEngine && cursorEngine != null && cursorSendLoop != null
-                        && !cursorSendLoop.delegateEngineClose()) {
-                    try {
-                        cursorEngine.close();
-                    } catch (Throwable t) {
-                        LOG.error("Error closing owned CursorSendEngine: {}", String.valueOf(t));
-                        terminalError = captureCloseError(terminalError, t);
+                if (ownsCursorEngine && cursorEngine != null && cursorSendLoop != null) {
+                    if (cursorSendLoop.delegateEngineClose()) {
+                        cursorEngineCloseDelegated = true;
+                    } else {
+                        try {
+                            closeOwnedCursorEngine();
+                        } catch (Throwable t) {
+                            LOG.error("Error closing owned CursorSendEngine: {}", String.valueOf(t));
+                            terminalError = captureCloseError(terminalError, t);
+                        }
                     }
-                    cursorEngine = null;
-                    ownsCursorEngine = false;
-                    slotLockReleased = true;
                 }
                 rethrowTerminal(terminalError);
                 return;
@@ -1232,18 +1247,15 @@ public void close() {
 
             if (ownsCursorEngine && cursorEngine != null) {
                 try {
-                    cursorEngine.close();
+                    closeOwnedCursorEngine();
                 } catch (Throwable t) {
                     LOG.error("Error closing owned CursorSendEngine: {}", String.valueOf(t));
                     terminalError = captureCloseError(terminalError, t);
                 }
-                cursorEngine = null;
-                ownsCursorEngine = false;
+            } else {
+                // This sender never owned an engine holding the pool's slot.
+                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
@@ -1283,7 +1295,6 @@ public void close() {
                 terminalError = null;
             }
             rethrowTerminal(terminalError);
-        }
     }
 
     /**
@@ -1293,7 +1304,8 @@ public void close() {
      * slot index reserved instead of reusing the still-locked slot dir.
      */
     public boolean isSlotLockReleased() {
-        return slotLockReleased;
+        CursorSendEngine engine = cursorEngine;
+        return slotLockReleased || (engine != null && engine.isCloseCompleted());
     }
 
     @Override
@@ -2527,6 +2539,24 @@ public boolean wasEverConnected() {
         return l != null && l.hasEverConnected();
     }
 
+    private void closeOwnedCursorEngine() {
+        if (!ownsCursorEngine || cursorEngine == null || cursorEngineCloseDelegated) {
+            return;
+        }
+        cursorEngine.close();
+        if (cursorEngine.isCloseCompleted()) {
+            cursorEngine = null;
+            ownsCursorEngine = false;
+            slotLockReleased = true;
+        } else {
+            // Preserve the sole owner reference for a later retry. The pool
+            // observes false and retires this index rather than reusing a
+            // directory whose flock is still held.
+            slotLockReleased = false;
+            cursorEngine.closeEventually();
+        }
+    }
+
     private static Throwable captureCloseError(Throwable terminalError, Throwable t) {
         if (terminalError == null) {
             return t;

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

@@ -704,10 +704,15 @@ public void run() {
                 // between the failed close() and now: then it is safe (and
                 // necessary) to close the engine here.
                 if (ioThreadStopped || !loop.delegateEngineClose()) {
+                    // Make one bounded attempt. If manager quiescence times
+                    // out, transfer ownership so this pool task can terminate.
                     try {
-                        // engine.close() releases the slot lock too.
                         engine.close();
                     } catch (Throwable ignored) {
+                        // The deferred owner retries below when needed.
+                    }
+                    if (!engine.isCloseCompleted()) {
+                        engine.closeEventually();
                     }
                 } else {
                     LOG.warn("drainer slot {}: engine close delegated to the I/O thread; "

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

@@ -61,6 +61,12 @@ public final class CursorSendEngine implements QuietCloseable {
      * Default deadline for {@link #appendBlocking}: 30 seconds.
      */
     public static final long DEFAULT_APPEND_DEADLINE_NANOS = 30_000_000_000L;
+    private static final java.util.concurrent.ExecutorService DEFERRED_CLOSE_EXECUTOR =
+            java.util.concurrent.Executors.newCachedThreadPool(runnable -> {
+                Thread thread = new Thread(runnable, "qdb-cursor-engine-close");
+                thread.setDaemon(true);
+                return thread;
+            });
     private static final org.slf4j.Logger LOG =
             org.slf4j.LoggerFactory.getLogger(CursorSendEngine.class);
     private final long appendDeadlineNanos;
@@ -118,8 +124,11 @@ public final class CursorSendEngine implements QuietCloseable {
     // 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.
+    // the synchronized close() method and isCloseCompleted() accessor.
     private boolean closeCompleted;
+    // True once a dedicated daemon has accepted ownership of incomplete close
+    // retries. Guarded by synchronized closeEventually().
+    private boolean deferredCloseScheduled;
     // Producer-thread-only: timestamp of the last "we're backpressured" log
     // line, used to throttle. Plain long is fine.
     private long lastBackpressureLogNs;
@@ -613,6 +622,25 @@ public synchronized void close() {
         }
     }
 
+    /**
+     * Transfers an incomplete close to a dedicated daemon owner. This lets
+     * I/O and drainer executors terminate even when a manager service pass is
+     * permanently stalled. The deferred owner retains this engine, and thus
+     * its flock and native mappings, until a retry completes.
+     */
+    public synchronized void closeEventually() {
+        if (closeCompleted || deferredCloseScheduled) {
+            return;
+        }
+        deferredCloseScheduled = true;
+        try {
+            DEFERRED_CLOSE_EXECUTOR.execute(this::runDeferredClose);
+        } catch (Throwable t) {
+            deferredCloseScheduled = false;
+            throw t;
+        }
+    }
+
     /**
      * Pass-through to {@link SegmentRing#findSegmentContaining(long)}.
      */
@@ -637,6 +665,16 @@ public long getTotalBackpressureStalls() {
         return backpressureStallCount.get();
     }
 
+    /**
+     * True only after close released every engine-owned resource, including
+     * the SF slot flock. A false result means close deliberately retained the
+     * ring/watermark/lock because the manager worker did not quiesce; owners
+     * must preserve this engine and retry later.
+     */
+    public synchronized boolean isCloseCompleted() {
+        return closeCompleted;
+    }
+
     /**
      * Pass-through to {@link SegmentRing#nextSealedAfter(MmapSegment)}.
      */
@@ -718,6 +756,23 @@ public long recoveredOrphanTipFsn() {
      * Best-effort: logs and continues on failures, since we're already on
      * the close path.
      */
+    private void runDeferredClose() {
+        while (!isCloseCompleted()) {
+            try {
+                close();
+            } catch (Throwable ignored) {
+                // Retain ownership and retry after a bounded pause.
+            }
+            if (!isCloseCompleted()) {
+                LockSupport.parkNanos(10_000_000L);
+                // The internal daemon owns cleanup independently of caller
+                // cancellation. Clear interrupts so they cannot create a hot
+                // retry loop if an executor implementation interrupts it.
+                Thread.interrupted();
+            }
+        }
+    }
+
     private static void unlinkAllSegmentFiles(String dir) {
         if (!io.questdb.client.std.Files.exists(dir)) return;
         long find = io.questdb.client.std.Files.findFirst(dir);

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

@@ -1679,10 +1679,16 @@ private void ioLoop() {
             // either this branch or the owner's fallback runs (or both —
             // engine.close() is idempotent).
             if (engineCloseDelegated) {
+                // Make one bounded attempt on this exit path. If manager
+                // quiescence times out, transfer ownership to the dedicated
+                // deferred closer so this I/O daemon can terminate.
                 try {
                     engine.close();
                 } catch (Throwable ignored) {
-                    // best-effort
+                    // The deferred owner retries below when needed.
+                }
+                if (!engine.isCloseCompleted()) {
+                    engine.closeEventually();
                 }
             }
         }