fix(qwp): recover SF slots after deferred cleanup

Author: bluestreak01

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/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/java/io/questdb/client/cutlass/qwp/client/QwpWebSocketSender.java

@@ -194,6 +194,9 @@ public class QwpWebSocketSender implements Sender {
     // 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
@@ -1057,6 +1060,16 @@ public QwpWebSocketSender charColumn(CharSequence columnName, char value) {
     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).
@@ -1275,10 +1288,10 @@ public void close() {
                     // 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 the worker's exit path
-                    // (owned manager); the getter re-probes the retained
-                    // engine so the pool can reclaim the slot once cleanup
-                    // actually completes.
+                    // 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;
                 }
@@ -1335,8 +1348,8 @@ public void close() {
      * index reserved instead of reusing the still-locked dir.
      * <p>
      * Not a one-shot snapshot: when close() left engine cleanup pending on a
-     * worker/I/O-thread exit path, this re-probes the retained engine and
-     * latches true the moment that cleanup completes — pools re-probe retired
+     * 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.
@@ -2194,6 +2207,16 @@ public void setClientFactoryOverride(java.util.function.Supplier<WebSocketClient
         this.clientFactoryOverride = factory;
     }
 
+    /**
+     * 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();
@@ -2262,7 +2285,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) {
@@ -2272,6 +2297,9 @@ 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;
     }