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

Author: bluestreak01

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);
@@ -354,3 +388,15 @@ JNIEXPORT jint JNICALL Java_io_questdb_client_std_Files_msync
         (JNIEnv *e, jclass cl, jlong addr, jlong len, jboolean async) {
     return msync((void *) (uintptr_t) addr, (size_t) len, async ? MS_ASYNC : MS_SYNC);
 }
+
+/* Best-effort page pin. Callers treat a refusal (RLIMIT_MEMLOCK, missing
+ * privilege) as a soft downgrade, so no errno capture is required here. */
+JNIEXPORT jint JNICALL Java_io_questdb_client_std_Files_mlock0
+        (JNIEnv *e, jclass cl, jlong addr, jlong len) {
+    return mlock((void *) (uintptr_t) addr, (size_t) len);
+}
+
+JNIEXPORT jint JNICALL Java_io_questdb_client_std_Files_munlock0
+        (JNIEnv *e, jclass cl, jlong addr, jlong len) {
+    return munlock((void *) (uintptr_t) addr, (size_t) len);
+}

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/share/net.h

@@ -6,4 +6,10 @@ static DWORD dwTlsIndexLastError = 0;
 
 void SaveLastError();
 
+/* Returns the per-thread error code most recently stored by SaveLastError().
+ * Both functions are defined in os.c, the only translation unit whose static
+ * dwTlsIndexLastError copy is initialised by TlsAlloc() in DllMain; every
+ * other includer's copy of the variable is an unused zero. */
+DWORD GetSavedLastError();
+
 #endif //ZLIB_ERRNO_H

core/src/main/c/windows/errno.h

@@ -77,11 +77,16 @@ static jint open_file(const char *utf8Path,
     }
     HANDLE h = CreateFileW(wide, desiredAccess, shareMode, NULL,
                            creationDisposition, flagsAndAttributes, NULL);
-    free(wide);
     if (h == INVALID_HANDLE_VALUE) {
+        // Save the CreateFileW failure code BEFORE free(): the thread's
+        // last-error value may be overwritten by any subsequent API/CRT call
+        // (HeapFree can SetLastError), and callers such as fsyncDir0 rely on
+        // the saved value being the CreateFileW error.
         SaveLastError();
+        free(wide);
         return -1;
     }
+    free(wide);
     return HANDLE_TO_FD(h);
 }
 
@@ -103,6 +108,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 +229,52 @@ 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. Consult the error open_file() saved at the CreateFileW failure
+        // point rather than the live GetLastError(): the intervening free()
+        // and return path inside open_file() are not guaranteed to preserve
+        // the thread's last-error value. A genuine failure such as a missing
+        // directory (ERROR_PATH_NOT_FOUND) still propagates as fatal.
+        if (GetSavedLastError() == 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;
@@ -300,18 +360,24 @@ JNIEXPORT jlong JNICALL Java_io_questdb_client_std_Files_length0
                            OPEN_EXISTING,
                            FILE_ATTRIBUTE_NORMAL,
                            NULL);
-    free(wide);
     if (h == INVALID_HANDLE_VALUE) {
+        /* Save the CreateFileW failure code BEFORE free(): any subsequent
+         * API/CRT call (HeapFree can SetLastError) may overwrite the
+         * thread's last-error value -- same pattern as open_file(). */
         SaveLastError();
+        free(wide);
         return -1;
     }
+    free(wide);
     LARGE_INTEGER sz;
     BOOL ok = GetFileSizeEx(h, &sz);
-    CloseHandle(h);
     if (!ok) {
+        /* Save before CloseHandle() can clobber the last-error value. */
         SaveLastError();
+        CloseHandle(h);
         return -1;
     }
+    CloseHandle(h);
     return (jlong) sz.QuadPart;
 }
 
@@ -331,6 +397,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;
@@ -340,11 +420,13 @@ JNIEXPORT jint JNICALL Java_io_questdb_client_std_Files_mkdir0
         return -1;
     }
     BOOL ok = CreateDirectoryW(wide, NULL);
-    free(wide);
     if (!ok) {
+        /* Save before free() can clobber the last-error value. */
         SaveLastError();
+        free(wide);
         return -1;
     }
+    free(wide);
     return 0;
 }
 
@@ -373,11 +455,13 @@ JNIEXPORT jboolean JNICALL Java_io_questdb_client_std_Files_remove0
     } else {
         ok = DeleteFileW(wide);
     }
-    free(wide);
     if (!ok) {
+        /* Save before free() can clobber the last-error value. */
         SaveLastError();
+        free(wide);
         return JNI_FALSE;
     }
+    free(wide);
     return JNI_TRUE;
 }
 
@@ -395,12 +479,15 @@ JNIEXPORT jint JNICALL Java_io_questdb_client_std_Files_rename0
         return -1;
     }
     BOOL ok = MoveFileExW(oldW, newW, MOVEFILE_REPLACE_EXISTING);
-    free(oldW);
-    free(newW);
     if (!ok) {
+        /* Save before free() can clobber the last-error value. */
         SaveLastError();
+        free(oldW);
+        free(newW);
         return -1;
     }
+    free(oldW);
+    free(newW);
     return 0;
 }
 
@@ -438,24 +525,28 @@ JNIEXPORT jlong JNICALL Java_io_questdb_client_std_Files_findFirst0
     pattern[pathLen] = '\0';
 
     wchar_t *wide = utf8_to_wide(pattern);
-    free(pattern);
     if (!wide) {
+        /* Save before free() can clobber the last-error value. */
         SaveLastError();
+        free(pattern);
         return 0;
     }
+    free(pattern);
 
     qdb_find_t *find = (qdb_find_t *) malloc(sizeof(qdb_find_t));
     if (!find) {
         free(wide);
         return 0;
     }
     find->handle = FindFirstFileW(wide, &find->data);
-    free(wide);
     if (find->handle == INVALID_HANDLE_VALUE) {
+        /* Save before free() can clobber the last-error value. */
         SaveLastError();
+        free(wide);
         free(find);
         return 0;
     }
+    free(wide);
     find->hasEntry = 1;
     win_findname_to_utf8(find);
     return (jlong) (uintptr_t) find;
@@ -615,3 +706,16 @@ JNIEXPORT jint JNICALL Java_io_questdb_client_std_Files_msync
     (void) async;
     return 0;
 }
+
+/* Best-effort page pin. VirtualLock is quota-bound to the process working-set
+ * minimum; callers treat any refusal as a soft downgrade, so no
+ * SaveLastError() -- the refusal is not surfaced as an error. */
+JNIEXPORT jint JNICALL Java_io_questdb_client_std_Files_mlock0
+        (JNIEnv *e, jclass cl, jlong addr, jlong len) {
+    return VirtualLock((LPVOID) (uintptr_t) addr, (SIZE_T) len) ? 0 : -1;
+}
+
+JNIEXPORT jint JNICALL Java_io_questdb_client_std_Files_munlock0
+        (JNIEnv *e, jclass cl, jlong addr, jlong len) {
+    return VirtualUnlock((LPVOID) (uintptr_t) addr, (SIZE_T) len) ? 0 : -1;
+}