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

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

@@ -124,6 +124,21 @@ JNIEXPORT jint JNICALL Java_io_questdb_client_std_Files_fsync
     return res;
 }
 
+JNIEXPORT jint JNICALL Java_io_questdb_client_std_Files_fsyncDir0
+        (JNIEnv *e, jclass cl, jlong lpszName) {
+    int fd;
+    RESTARTABLE(open((const char *) (uintptr_t) lpszName, O_RDONLY), fd);
+    if (fd < 0) {
+        return -1;
+    }
+    int res;
+    RESTARTABLE(fsync(fd), res);
+    int saved_errno = errno;
+    close(fd);
+    errno = saved_errno;
+    return res;
+}
+
 JNIEXPORT jboolean JNICALL Java_io_questdb_client_std_Files_truncate
         (JNIEnv *e, jclass cl, jint fd, jlong size) {
     int res;
@@ -236,6 +251,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,12 @@ 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) {
+    const int result = shutdown((int) fd, SHUT_RDWR);
+    return result == -1 && errno == ENOTCONN ? 0 : result;
+}
+
 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

@@ -215,6 +215,25 @@ JNIEXPORT jint JNICALL Java_io_questdb_client_std_Files_fsync
     return 0;
 }
 
+JNIEXPORT jint JNICALL Java_io_questdb_client_std_Files_fsyncDir0
+        (JNIEnv *e, jclass cl, jlong lpszName) {
+    jint fd = open_file((const char *) (uintptr_t) lpszName,
+                        GENERIC_READ | GENERIC_WRITE,
+                        FILE_SHARE_READ | FILE_SHARE_WRITE | FILE_SHARE_DELETE,
+                        OPEN_EXISTING,
+                        FILE_FLAG_BACKUP_SEMANTICS);
+    if (fd < 0) {
+        return -1;
+    }
+    if (!FlushFileBuffers(FD_TO_HANDLE(fd))) {
+        SaveLastError();
+        CloseHandle(FD_TO_HANDLE(fd));
+        return -1;
+    }
+    CloseHandle(FD_TO_HANDLE(fd));
+    return 0;
+}
+
 JNIEXPORT jboolean JNICALL Java_io_questdb_client_std_Files_truncate
         (JNIEnv *e, jclass cl, jint fd, jlong size) {
     FILE_END_OF_FILE_INFO eof;
@@ -331,6 +350,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,20 @@ 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) {
+        const int error = WSAGetLastError();
+        if (error == WSAENOTCONN) {
+            return 0;
+        }
+        WSASetLastError(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.
      *