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

README.md

@@ -142,8 +142,9 @@ You can also let the client flush batches for you with the `auto_flush_rows` / `
 `ws::addr=localhost:9000;auto_flush_rows=10000;auto_flush_interval=1000;`.
 
 **Confirm a batch is durably received.** Over QWP each flush returns a frame sequence number (FSN); `awaitAckedFsn`
-blocks until the server has acknowledged it. Rows are safe in the store-and-forward log even if the ack has not landed
-yet — they replay on reconnect.
+blocks until the server has acknowledged it. With `sf_dir`, rows in the store-and-forward log replay after reconnect
+or a producer-process restart. For periodic host-power-loss checkpoints, also configure
+`sf_durability=periodic;sf_sync_interval_millis=5000;`.
 
 ```java
 try (Sender sender = db.borrowSender()) {
@@ -259,7 +260,9 @@ try (QuestDB db = QuestDB.builder()
 List every cluster node in one `addr` server list; the single string configures both the ingest and query pools across
 all of them. On the query side, `target` selects the node role to route to (`any`, `primary`, or `replica`) and
 `failover=on` enables failover across the list. The ingest side reconnects across the same node list on its own — a
-store-and-forward sender keeps buffering rows through a failover window and never drops them.
+store-and-forward sender keeps buffering rows through a failover window and replays unacknowledged data. The default
+`memory` durability mode protects process restarts, not host power loss; use periodic durability for background disk
+checkpoints.
 
 ```java
 try (QuestDB db = QuestDB.connect(
@@ -439,7 +442,10 @@ Applied by the query pool to select and fail over between the nodes in the `addr
 | `client_id` |         | Opaque client identifier surfaced server-side for observability                          |
 
 The ingest side also accepts store-and-forward and reconnection tuning keys (`auto_flush_*`, `initial_connect_retry`,
-`reconnect_*`, `request_durable_ack`, `sf_*`, `max_frame_rejections`, `poison_min_escalation_window_millis`, …). See the
+`reconnect_*`, `request_durable_ack`, `sf_*`, `max_frame_rejections`, `poison_min_escalation_window_millis`, …).
+`sf_durability=periodic` checkpoints mmap-published data in the background; `sf_sync_interval_millis` defaults to `5000`
+in that mode. The interval is a target cadence: JVM scheduling and storage-sync latency add to the actual loss window.
+Use `request_durable_ack=on` when end-to-end server durability is also required. See the
 [QuestDB documentation](https://questdb.com/docs/) for the full reference.
 
 ## Requirements

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

@@ -65,6 +65,13 @@ JNIEXPORT jint JNICALL Java_io_questdb_client_std_Files_openRW0
     return (jint) fd;
 }
 
+JNIEXPORT jint JNICALL Java_io_questdb_client_std_Files_openRWExclusive0
+        (JNIEnv *e, jclass cl, jlong lpszName) {
+    int fd;
+    RESTARTABLE(open((const char *) (uintptr_t) lpszName, O_CREAT | O_EXCL | O_RDWR, 0644), fd);
+    return (jint) fd;
+}
+
 JNIEXPORT jint JNICALL Java_io_questdb_client_std_Files_openAppend0
         (JNIEnv *e, jclass cl, jlong lpszName) {
     int fd;
@@ -124,6 +131,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 +258,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

@@ -103,6 +103,15 @@ JNIEXPORT jint JNICALL Java_io_questdb_client_std_Files_openRW0
                      FILE_ATTRIBUTE_NORMAL);
 }
 
+JNIEXPORT jint JNICALL Java_io_questdb_client_std_Files_openRWExclusive0
+        (JNIEnv *e, jclass cl, jlong lpszName) {
+    return open_file((const char *) (uintptr_t) lpszName,
+                     GENERIC_READ | GENERIC_WRITE,
+                     FILE_SHARE_READ | FILE_SHARE_WRITE | FILE_SHARE_DELETE,
+                     CREATE_NEW,
+                     FILE_ATTRIBUTE_NORMAL);
+}
+
 JNIEXPORT jint JNICALL Java_io_questdb_client_std_Files_openAppend0
         (JNIEnv *e, jclass cl, jlong lpszName) {
     jint fd = open_file((const char *) (uintptr_t) lpszName,
@@ -215,6 +224,50 @@ 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) {
+        // A directory can be crash-consistent yet not openable for write:
+        // some ACL/filesystem configurations refuse a GENERIC_WRITE open of a
+        // directory handle with ERROR_ACCESS_DENIED. Directory-entry durability
+        // on NTFS is provided by metadata journaling ($LogFile), so degrade to
+        // best-effort success here rather than hard-failing the SF durability
+        // path. open_file() already recorded the error via SaveLastError(); a
+        // genuine failure such as a missing directory (ERROR_PATH_NOT_FOUND)
+        // still propagates as fatal.
+        if (GetLastError() == ERROR_ACCESS_DENIED) {
+            return 0;
+        }
+        return -1;
+    }
+    if (!FlushFileBuffers(FD_TO_HANDLE(fd))) {
+        DWORD err = GetLastError();
+        CloseHandle(FD_TO_HANDLE(fd));
+        // FlushFileBuffers is documented for file/volume handles; NTFS refuses
+        // it on a directory handle (typically ERROR_ACCESS_DENIED, and
+        // ERROR_INVALID_FUNCTION on filesystems that do not implement it).
+        // create/rename/unlink of directory entries are made crash consistent
+        // by NTFS metadata journaling, so treat those "not supported"
+        // signatures as best-effort success. This mirrors libgit2/PostgreSQL/
+        // SQLite, which do not rely on directory fsync on Windows, and keeps
+        // the SF manifest-create, slot-lock, and trim/unlink barriers from
+        // hard-failing. Any other error is a real I/O failure and stays fatal.
+        if (err == ERROR_ACCESS_DENIED || err == ERROR_INVALID_FUNCTION) {
+            return 0;
+        }
+        SetLastError(err);
+        SaveLastError();
+        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 +384,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,28 @@ 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) {
+    // Unlike POSIX, a Winsock shutdown() does not unblock a recv() that is
+    // already in progress on another thread. Cancel any outstanding I/O on the
+    // handle first so a worker parked in a blocking recv() wakes up. CancelIoEx
+    // does not close the socket, so the descriptor stays allocated and the fd
+    // cannot be reused underneath the worker before close() releases it. A
+    // FALSE return with ERROR_NOT_FOUND (nothing was pending) is expected here
+    // and deliberately ignored.
+    CancelIoEx((HANDLE) (SOCKET) fd, NULL);
+    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/QuestDB.java

@@ -130,10 +130,17 @@ static QuestDB connect(CharSequence configurationString) {
     Sender borrowSender();
 
     /**
-     * Shuts down the pools, closing every underlying {@link Sender} and
-     * query client. Idempotent. Threads currently blocked in
-     * {@link #borrowSender()} or {@link Query#submit()} are released with an
-     * error.
+     * Shuts down the pools and their published clients. Idempotent. Threads
+     * currently blocked in {@link #borrowSender()} or {@link Query#submit()}
+     * are released with an error.
+     * <p>
+     * In-progress client creation: close() waits up to the builder's
+     * {@link QuestDBBuilder#acquireTimeoutMillis(long) acquire timeout},
+     * hard-capped at 5 seconds, for a creation blocked in DNS, TCP, TLS, or a
+     * WebSocket handshake. If that budget expires, the creator retains cleanup
+     * ownership and closes its unpublished client (and releases any SF slot)
+     * when construction returns. This keeps close() bounded even when
+     * {@code connect_timeout} is unset without abandoning late resources.
      * <p>
      * Outstanding leases: a borrowed {@link Sender} is never torn down
      * underneath the thread using it. Instead, close() waits up to the