fix(qwp): keep SF slot locked until manager worker quiesces

core/src/main/c/share/files.c

@@ -236,6 +236,18 @@ JNIEXPORT jint JNICALL Java_io_questdb_client_std_Files_lock
     return flock((int) fd, LOCK_EX | LOCK_NB);
 }
 
+JNIEXPORT jint JNICALL Java_io_questdb_client_cutlass_qwp_client_sf_cursor_SlotLock_release0
+        (JNIEnv *e, jclass cl, jint fd) {
+    if (flock((int) fd, LOCK_UN) != 0) {
+        return -1;
+    }
+    /* Unlock success confirms that the slot is reusable. close() is one-shot:
+     * POSIX leaves descriptor state unspecified on EINTR, so retrying its
+     * numeric value could close an unrelated descriptor after reuse. */
+    (void) close((int) fd);
+    return 0;
+}
+
 JNIEXPORT jint JNICALL Java_io_questdb_client_std_Files_mkdir0
         (JNIEnv *e, jclass cl, jlong lpszPath, jint mode) {
     return mkdir((const char *) (uintptr_t) lpszPath, (mode_t) mode);

core/src/main/c/share/net.c

@@ -128,6 +128,11 @@ JNIEXPORT jint JNICALL Java_io_questdb_client_network_Net_send
     return com_questdb_network_Net_EOTHERDISCONNECT;
 }
 
+JNIEXPORT jint JNICALL Java_io_questdb_client_network_Net_shutdown
+        (JNIEnv *e, jclass cl, jint fd) {
+    return shutdown((int) fd, SHUT_RDWR);
+}
+
 JNIEXPORT jint JNICALL Java_io_questdb_client_network_Net_recv
         (JNIEnv *e, jclass cl, jint fd, jlong ptr, jint len) {
     ssize_t n;

core/src/main/c/windows/files.c

@@ -331,6 +331,20 @@ JNIEXPORT jint JNICALL Java_io_questdb_client_std_Files_lock
     return 0;
 }
 
+JNIEXPORT jint JNICALL Java_io_questdb_client_cutlass_qwp_client_sf_cursor_SlotLock_release0
+        (JNIEnv *e, jclass cl, jint fd) {
+    OVERLAPPED ov;
+    memset(&ov, 0, sizeof(ov));
+    if (!UnlockFileEx(FD_TO_HANDLE(fd), 0, MAXDWORD, MAXDWORD, &ov)) {
+        SaveLastError();
+        return -1;
+    }
+    /* Unlock success confirms that the slot is reusable. Match POSIX by
+     * closing once without making handle cleanup part of that signal. */
+    (void) CloseHandle(FD_TO_HANDLE(fd));
+    return 0;
+}
+
 JNIEXPORT jint JNICALL Java_io_questdb_client_std_Files_mkdir0
         (JNIEnv *e, jclass cl, jlong lpszPath, jint mode) {
     (void) mode;

core/src/main/c/windows/net.c

@@ -230,6 +230,15 @@ JNIEXPORT jint JNICALL Java_io_questdb_client_network_Net_configureNonBlocking
     return res;
 }
 
+JNIEXPORT jint JNICALL Java_io_questdb_client_network_Net_shutdown
+        (JNIEnv *e, jclass cl, jint fd) {
+    const int result = shutdown((SOCKET) fd, SD_BOTH);
+    if (result == SOCKET_ERROR) {
+        SaveLastError();
+    }
+    return result;
+}
+
 JNIEXPORT jint JNICALL Java_io_questdb_client_network_Net_recv
         (JNIEnv *e, jclass cl, jint fd, jlong addr, jint len) {
     const int n = recv((SOCKET) fd, (char *) addr, len, 0);

core/src/main/java/io/questdb/client/cutlass/http/client/WebSocketClient.java

@@ -248,6 +248,15 @@ public void close() {
         }
     }
 
+    /**
+     * Shuts down socket traffic without releasing the descriptor or freeing
+     * buffers that a concurrent I/O worker may still access. The owner must
+     * call {@link #close()} after joining the worker to complete cleanup.
+     */
+    public void closeTraffic() {
+        socket.closeTraffic();
+    }
+
     /**
      * Connects to a WebSocket server.
      *

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

@@ -189,11 +189,17 @@ public class QwpWebSocketSender implements Sender {
     // Null in production; set reflectively by tests.
     @TestOnly
     private volatile java.util.function.Supplier<WebSocketClient> clientFactoryOverride;
+    // Test-only lifecycle witness. drainOnClose() invokes and clears it after
+    // confirming a real unacknowledged close target, immediately before waiting.
+    private volatile Runnable closeDrainWaitingHook;
     // close() drain timeout in millis. Default applied at construction.
     // 0 or -1 means "fast close" (skip the drain); otherwise close blocks
     // up to this many millis for ackedFsn to catch up to publishedFsn.
     private long closeFlushTimeoutMillis = 5_000L;
     private volatile boolean closed;
+    // Test-only lifecycle witness. close() invokes and clears it strictly after
+    // publishing closed=true and before starting any drain or teardown work.
+    private volatile Runnable closeStartedHook;
     private boolean connected;
     private SenderConnectionDispatcher connectionDispatcher;
     // Async-delivery sink for SenderConnectionEvent notifications. Default
@@ -278,11 +284,23 @@ 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").
-    private boolean slotLockReleased;
+    // 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"). May flip to true LATER via the getter's
+    // re-probe of retainedEngine, once the deferred cleanup (manager-worker
+    // exit path or delegated I/O-thread close) releases the flock — pools
+    // re-probe retired slots to recover their capacity. volatile: written on
+    // the closing thread, read by pool threads.
+    private volatile boolean slotLockReleased;
+    // Optional owning-pool notification, relayed after the engine confirms the
+    // flock release and slotLockReleased is visible to pool re-probes.
+    private volatile Runnable slotLockReleaseListener;
+    // Engine whose close() could not complete during sender close() — its
+    // cleanup is pending on a worker/I/O-thread exit path. isSlotLockReleased()
+    // re-probes it so a late flock release becomes visible to the owning pool.
+    // Only ever set inside close(); null for a sender that closed cleanly.
+    private volatile CursorSendEngine retainedEngine;
     private int pendingRowCount;
     private SenderProgressDispatcher progressDispatcher;
     // Async-delivery sink for ack-watermark advances. Default no-op; a
@@ -1038,13 +1056,26 @@ 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
     public void close() {
         if (!closed) {
             closed = true;
+            Runnable hook = closeStartedHook;
+            closeStartedHook = null;
+            if (hook != null) {
+                try {
+                    hook.run();
+                } catch (Throwable t) {
+                    // A test witness must never prevent production resource cleanup.
+                    LOG.error("Error in close-started test hook: {}", String.valueOf(t));
+                }
+            }
             boolean ioThreadStopped = true;
             // Captures the first error from the flush/drain path AND any
             // secondary errors from cleanup steps (added via addSuppressed).
@@ -1187,17 +1218,34 @@ 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()) {
+                        // The I/O thread adopted the engine close and runs it on
+                        // its exit path. Keep the engine visible for re-probing:
+                        // isSlotLockReleased() flips true once that close
+                        // completes, letting a pool recover the retired slot.
+                        retainedEngine = cursorEngine;
+                    } else {
+                        CursorSendEngine engine = cursorEngine;
+                        try {
+                            engine.close();
+                        } catch (Throwable t) {
+                            LOG.error("Error closing owned CursorSendEngine: {}", String.valueOf(t));
+                            terminalError = captureCloseError(terminalError, t);
+                        }
+                        if (engine.isCloseCompleted()) {
+                            cursorEngine = null;
+                            ownsCursorEngine = false;
+                            slotLockReleased = true;
+                        } else {
+                            // Engine cleanup is pending on its manager worker's
+                            // exit path (or leaked, for a shared manager). Report
+                            // the retained flock now; the getter re-probes the
+                            // retained engine so a late release is not lost.
+                            slotLockReleased = false;
+                            retainedEngine = engine;
+                        }
                     }
-                    cursorEngine = null;
-                    ownsCursorEngine = false;
-                    slotLockReleased = true;
                 }
                 rethrowTerminal(terminalError);
                 return;
@@ -1231,19 +1279,32 @@ 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.
+                    // Engine cleanup was handed to a safe manager-worker path:
+                    // owned-manager exit or shared-manager ring-pass completion.
+                    // The getter re-probes the retained engine so the pool can
+                    // reclaim the slot once cleanup actually completes.
+                    slotLockReleased = false;
+                    retainedEngine = engine;
+                }
+            } 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
@@ -1287,13 +1348,41 @@ 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.
+     * True once the store-and-forward slot flock has been released. 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.
+     * <p>
+     * Not a one-shot snapshot: when close() left engine cleanup pending on a
+     * manager-worker quiescence or I/O-thread exit path, this re-probes the
+     * retained engine and latches true the moment that cleanup completes — pools re-probe retired
+     * slots through this getter to recover their capacity. Monotonic:
+     * false→true only, never back. Lock-free (volatile reads) so pools may
+     * call it under their capacity lock.
      */
     public boolean isSlotLockReleased() {
-        return slotLockReleased;
+        if (slotLockReleased) {
+            return true;
+        }
+        CursorSendEngine engine = retainedEngine;
+        if (engine != null && engine.isCloseCompleted()) {
+            // Benign latch race: concurrent callers may both observe the
+            // completed cleanup and both write true.
+            slotLockReleased = true;
+            return true;
+        }
+        return false;
+    }
+
+    /**
+     * Registers a callback for confirmed SF slot-lock release. Pools use this
+     * to wake borrowers without waiting for a timeout or housekeeper tick.
+     */
+    public void setSlotLockReleaseListener(Runnable listener) {
+        slotLockReleaseListener = listener;
+        if (listener != null && slotLockReleased) {
+            listener.run();
+        }
     }
 
     @Override
@@ -2135,6 +2224,26 @@ public void setClientFactoryOverride(java.util.function.Supplier<WebSocketClient
         this.clientFactoryOverride = factory;
     }
 
+    /**
+     * Installs a one-shot test witness that close-time drain invokes after it
+     * observes a real unacknowledged target and before it starts waiting.
+     * Production code never sets it.
+     */
+    @TestOnly
+    public void setCloseDrainWaitingHook(Runnable hook) {
+        this.closeDrainWaitingHook = hook;
+    }
+
+    /**
+     * Installs a one-shot test witness that {@link #close()} invokes after it
+     * publishes the closed-state transition and before it starts drain or
+     * teardown work. Production code never sets it.
+     */
+    @TestOnly
+    public void setCloseStartedHook(Runnable hook) {
+        this.closeStartedHook = hook;
+    }
+
     @Override
     public void reset() {
         checkNotClosed();
@@ -2203,7 +2312,9 @@ public void setConnectionListenerInboxCapacity(int capacity) {
 
     /**
      * Attach a {@link CursorSendEngine} for store-and-forward. Must be called
-     * before the first send.
+     * before the first send. Once a non-null engine has been attached, it
+     * cannot be replaced or detached. Ownership of a rejected engine remains
+     * with the caller.
      */
     public void setCursorEngine(CursorSendEngine engine, boolean takeOwnership) {
         if (closed) {
@@ -2213,8 +2324,14 @@ public void setCursorEngine(CursorSendEngine engine, boolean takeOwnership) {
             throw new LineSenderException(
                     "setCursorEngine must be called before the first send");
         }
+        if (cursorEngine != null) {
+            throw new LineSenderException("CursorSendEngine is already attached");
+        }
         this.cursorEngine = engine;
         this.ownsCursorEngine = takeOwnership && engine != null;
+        if (engine != null) {
+            engine.setSlotLockReleaseListener(this::onSlotLockReleased);
+        }
     }
 
     /**
@@ -3185,6 +3302,16 @@ private void drainOnClose(boolean errorOwnedByCustomHandler) {
         if (cursorEngine.ackedFsn() >= target) {
             return;
         }
+        Runnable hook = closeDrainWaitingHook;
+        closeDrainWaitingHook = null;
+        if (hook != null) {
+            try {
+                hook.run();
+            } catch (Throwable t) {
+                // A test witness must never prevent production resource cleanup.
+                LOG.error("Error in close-drain-waiting test hook: {}", String.valueOf(t));
+            }
+        }
         long deadlineNanos = System.nanoTime() + closeFlushTimeoutMillis * 1_000_000L;
         while (cursorEngine.ackedFsn() < target) {
             // Stop on a latched terminal (acks will never reach target);
@@ -3545,6 +3672,14 @@ private void flushPendingRowsSplit(ObjList<CharSequence> keys, boolean deferComm
         resetTableBuffersAfterFlush(keys);
     }
 
+    private void onSlotLockReleased() {
+        slotLockReleased = true;
+        Runnable listener = slotLockReleaseListener;
+        if (listener != null) {
+            listener.run();
+        }
+    }
+
     private void resetTableBuffersAfterFlush(ObjList<CharSequence> keys) {
         for (int i = 0, n = keys.size(); i < n; i++) {
             CharSequence tableName = keys.getQuick(i);