Skip to content

Commit 2e873f0

Browse files
mtopolnikclaude
andcommitted
Merge main; reconcile sf rename with periodic sync
Merge main (PR #67, keep SF slot locked until manager worker quiesces, plus its periodic-durability work) into the branch that renames sf_max_bytes to sf_max_segment_bytes. The merge base is one commit behind on each side, so every conflict is this branch's rename landing on top of #67's edits; each was resolved by keeping main's semantic change and re-applying the rename. Conflict resolutions: - Sender.java (3 spots): kept main's reworded comment and its new actualSfSyncIntervalNanos argument threaded into the CursorSendEngine constructor and startOrphanDrainers, with the identifiers renamed to actualSfMaxSegmentBytes / sfMaxSegmentBytes. - SegmentRing.java: took main's version wholesale. The only line this branch changed here was a comment that #67's recover() rewrite deleted, so main's file already reflects the intended state. - SfFromConfigTest.java: kept main's four new testSfSyncInterval* methods and renamed the trailing method to testSfMaxSegmentBytesAcceptsSizeSuffixes. - SegmentRingTest.java: kept main's Javadoc wording (it matches the deterministic comparison-count assertion) and applied the rename. Verified no conflict markers or stale sf_max_bytes spellings remain, and that the merged call sites resolve against the five-argument CursorSendEngine constructor and startOrphanDrainers overload. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2 parents 872c182 + 5bbdfb8 commit 2e873f0

100 files changed

Lines changed: 23745 additions & 2050 deletions

File tree

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

README.md

Lines changed: 10 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -142,8 +142,9 @@ You can also let the client flush batches for you with the `auto_flush_rows` / `
142142
`ws::addr=localhost:9000;auto_flush_rows=10000;auto_flush_interval=1000;`.
143143

144144
**Confirm a batch is durably received.** Over QWP each flush returns a frame sequence number (FSN); `awaitAckedFsn`
145-
blocks until the server has acknowledged it. Rows are safe in the store-and-forward log even if the ack has not landed
146-
yet — they replay on reconnect.
145+
blocks until the server has acknowledged it. With `sf_dir`, rows in the store-and-forward log replay after reconnect
146+
or a producer-process restart. For periodic host-power-loss checkpoints, also configure
147+
`sf_durability=periodic;sf_sync_interval_millis=5000;`.
147148

148149
```java
149150
try (Sender sender = db.borrowSender()) {
@@ -259,7 +260,9 @@ try (QuestDB db = QuestDB.builder()
259260
List every cluster node in one `addr` server list; the single string configures both the ingest and query pools across
260261
all of them. On the query side, `target` selects the node role to route to (`any`, `primary`, or `replica`) and
261262
`failover=on` enables failover across the list. The ingest side reconnects across the same node list on its own — a
262-
store-and-forward sender keeps buffering rows through a failover window and never drops them.
263+
store-and-forward sender keeps buffering rows through a failover window and replays unacknowledged data. The default
264+
`memory` durability mode protects process restarts, not host power loss; use periodic durability for background disk
265+
checkpoints.
263266

264267
```java
265268
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
439442
| `client_id` | | Opaque client identifier surfaced server-side for observability |
440443

441444
The ingest side also accepts store-and-forward and reconnection tuning keys (`auto_flush_*`, `initial_connect_retry`,
442-
`reconnect_*`, `request_durable_ack`, `sf_*`, `max_frame_rejections`, `poison_min_escalation_window_millis`, …). See the
445+
`reconnect_*`, `request_durable_ack`, `sf_*`, `max_frame_rejections`, `poison_min_escalation_window_millis`, …).
446+
`sf_durability=periodic` checkpoints mmap-published data in the background; `sf_sync_interval_millis` defaults to `5000`
447+
in that mode. The interval is a target cadence: JVM scheduling and storage-sync latency add to the actual loss window.
448+
Use `request_durable_ack=on` when end-to-end server durability is also required. See the
443449
[QuestDB documentation](https://questdb.com/docs/) for the full reference.
444450

445451
## Requirements

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

Lines changed: 46 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -65,6 +65,13 @@ JNIEXPORT jint JNICALL Java_io_questdb_client_std_Files_openRW0
6565
return (jint) fd;
6666
}
6767

68+
JNIEXPORT jint JNICALL Java_io_questdb_client_std_Files_openRWExclusive0
69+
(JNIEnv *e, jclass cl, jlong lpszName) {
70+
int fd;
71+
RESTARTABLE(open((const char *) (uintptr_t) lpszName, O_CREAT | O_EXCL | O_RDWR, 0644), fd);
72+
return (jint) fd;
73+
}
74+
6875
JNIEXPORT jint JNICALL Java_io_questdb_client_std_Files_openAppend0
6976
(JNIEnv *e, jclass cl, jlong lpszName) {
7077
int fd;
@@ -124,6 +131,21 @@ JNIEXPORT jint JNICALL Java_io_questdb_client_std_Files_fsync
124131
return res;
125132
}
126133

134+
JNIEXPORT jint JNICALL Java_io_questdb_client_std_Files_fsyncDir0
135+
(JNIEnv *e, jclass cl, jlong lpszName) {
136+
int fd;
137+
RESTARTABLE(open((const char *) (uintptr_t) lpszName, O_RDONLY), fd);
138+
if (fd < 0) {
139+
return -1;
140+
}
141+
int res;
142+
RESTARTABLE(fsync(fd), res);
143+
int saved_errno = errno;
144+
close(fd);
145+
errno = saved_errno;
146+
return res;
147+
}
148+
127149
JNIEXPORT jboolean JNICALL Java_io_questdb_client_std_Files_truncate
128150
(JNIEnv *e, jclass cl, jint fd, jlong size) {
129151
int res;
@@ -236,6 +258,18 @@ JNIEXPORT jint JNICALL Java_io_questdb_client_std_Files_lock
236258
return flock((int) fd, LOCK_EX | LOCK_NB);
237259
}
238260

261+
JNIEXPORT jint JNICALL Java_io_questdb_client_cutlass_qwp_client_sf_cursor_SlotLock_release0
262+
(JNIEnv *e, jclass cl, jint fd) {
263+
if (flock((int) fd, LOCK_UN) != 0) {
264+
return -1;
265+
}
266+
/* Unlock success confirms that the slot is reusable. close() is one-shot:
267+
* POSIX leaves descriptor state unspecified on EINTR, so retrying its
268+
* numeric value could close an unrelated descriptor after reuse. */
269+
(void) close((int) fd);
270+
return 0;
271+
}
272+
239273
JNIEXPORT jint JNICALL Java_io_questdb_client_std_Files_mkdir0
240274
(JNIEnv *e, jclass cl, jlong lpszPath, jint mode) {
241275
return mkdir((const char *) (uintptr_t) lpszPath, (mode_t) mode);
@@ -354,3 +388,15 @@ JNIEXPORT jint JNICALL Java_io_questdb_client_std_Files_msync
354388
(JNIEnv *e, jclass cl, jlong addr, jlong len, jboolean async) {
355389
return msync((void *) (uintptr_t) addr, (size_t) len, async ? MS_ASYNC : MS_SYNC);
356390
}
391+
392+
/* Best-effort page pin. Callers treat a refusal (RLIMIT_MEMLOCK, missing
393+
* privilege) as a soft downgrade, so no errno capture is required here. */
394+
JNIEXPORT jint JNICALL Java_io_questdb_client_std_Files_mlock0
395+
(JNIEnv *e, jclass cl, jlong addr, jlong len) {
396+
return mlock((void *) (uintptr_t) addr, (size_t) len);
397+
}
398+
399+
JNIEXPORT jint JNICALL Java_io_questdb_client_std_Files_munlock0
400+
(JNIEnv *e, jclass cl, jlong addr, jlong len) {
401+
return munlock((void *) (uintptr_t) addr, (size_t) len);
402+
}

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

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -128,6 +128,12 @@ JNIEXPORT jint JNICALL Java_io_questdb_client_network_Net_send
128128
return com_questdb_network_Net_EOTHERDISCONNECT;
129129
}
130130

131+
JNIEXPORT jint JNICALL Java_io_questdb_client_network_Net_shutdown
132+
(JNIEnv *e, jclass cl, jint fd) {
133+
const int result = shutdown((int) fd, SHUT_RDWR);
134+
return result == -1 && errno == ENOTCONN ? 0 : result;
135+
}
136+
131137
JNIEXPORT jint JNICALL Java_io_questdb_client_network_Net_recv
132138
(JNIEnv *e, jclass cl, jint fd, jlong ptr, jint len) {
133139
ssize_t n;

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

Lines changed: 8 additions & 0 deletions
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

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

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -6,4 +6,10 @@ static DWORD dwTlsIndexLastError = 0;
66

77
void SaveLastError();
88

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

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

Lines changed: 113 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -77,11 +77,16 @@ static jint open_file(const char *utf8Path,
7777
}
7878
HANDLE h = CreateFileW(wide, desiredAccess, shareMode, NULL,
7979
creationDisposition, flagsAndAttributes, NULL);
80-
free(wide);
8180
if (h == INVALID_HANDLE_VALUE) {
81+
// Save the CreateFileW failure code BEFORE free(): the thread's
82+
// last-error value may be overwritten by any subsequent API/CRT call
83+
// (HeapFree can SetLastError), and callers such as fsyncDir0 rely on
84+
// the saved value being the CreateFileW error.
8285
SaveLastError();
86+
free(wide);
8387
return -1;
8488
}
89+
free(wide);
8590
return HANDLE_TO_FD(h);
8691
}
8792

@@ -103,6 +108,15 @@ JNIEXPORT jint JNICALL Java_io_questdb_client_std_Files_openRW0
103108
FILE_ATTRIBUTE_NORMAL);
104109
}
105110

111+
JNIEXPORT jint JNICALL Java_io_questdb_client_std_Files_openRWExclusive0
112+
(JNIEnv *e, jclass cl, jlong lpszName) {
113+
return open_file((const char *) (uintptr_t) lpszName,
114+
GENERIC_READ | GENERIC_WRITE,
115+
FILE_SHARE_READ | FILE_SHARE_WRITE | FILE_SHARE_DELETE,
116+
CREATE_NEW,
117+
FILE_ATTRIBUTE_NORMAL);
118+
}
119+
106120
JNIEXPORT jint JNICALL Java_io_questdb_client_std_Files_openAppend0
107121
(JNIEnv *e, jclass cl, jlong lpszName) {
108122
jint fd = open_file((const char *) (uintptr_t) lpszName,
@@ -215,6 +229,52 @@ JNIEXPORT jint JNICALL Java_io_questdb_client_std_Files_fsync
215229
return 0;
216230
}
217231

232+
JNIEXPORT jint JNICALL Java_io_questdb_client_std_Files_fsyncDir0
233+
(JNIEnv *e, jclass cl, jlong lpszName) {
234+
jint fd = open_file((const char *) (uintptr_t) lpszName,
235+
GENERIC_READ | GENERIC_WRITE,
236+
FILE_SHARE_READ | FILE_SHARE_WRITE | FILE_SHARE_DELETE,
237+
OPEN_EXISTING,
238+
FILE_FLAG_BACKUP_SEMANTICS);
239+
if (fd < 0) {
240+
// A directory can be crash-consistent yet not openable for write:
241+
// some ACL/filesystem configurations refuse a GENERIC_WRITE open of a
242+
// directory handle with ERROR_ACCESS_DENIED. Directory-entry durability
243+
// on NTFS is provided by metadata journaling ($LogFile), so degrade to
244+
// best-effort success here rather than hard-failing the SF durability
245+
// path. Consult the error open_file() saved at the CreateFileW failure
246+
// point rather than the live GetLastError(): the intervening free()
247+
// and return path inside open_file() are not guaranteed to preserve
248+
// the thread's last-error value. A genuine failure such as a missing
249+
// directory (ERROR_PATH_NOT_FOUND) still propagates as fatal.
250+
if (GetSavedLastError() == ERROR_ACCESS_DENIED) {
251+
return 0;
252+
}
253+
return -1;
254+
}
255+
if (!FlushFileBuffers(FD_TO_HANDLE(fd))) {
256+
DWORD err = GetLastError();
257+
CloseHandle(FD_TO_HANDLE(fd));
258+
// FlushFileBuffers is documented for file/volume handles; NTFS refuses
259+
// it on a directory handle (typically ERROR_ACCESS_DENIED, and
260+
// ERROR_INVALID_FUNCTION on filesystems that do not implement it).
261+
// create/rename/unlink of directory entries are made crash consistent
262+
// by NTFS metadata journaling, so treat those "not supported"
263+
// signatures as best-effort success. This mirrors libgit2/PostgreSQL/
264+
// SQLite, which do not rely on directory fsync on Windows, and keeps
265+
// the SF manifest-create, slot-lock, and trim/unlink barriers from
266+
// hard-failing. Any other error is a real I/O failure and stays fatal.
267+
if (err == ERROR_ACCESS_DENIED || err == ERROR_INVALID_FUNCTION) {
268+
return 0;
269+
}
270+
SetLastError(err);
271+
SaveLastError();
272+
return -1;
273+
}
274+
CloseHandle(FD_TO_HANDLE(fd));
275+
return 0;
276+
}
277+
218278
JNIEXPORT jboolean JNICALL Java_io_questdb_client_std_Files_truncate
219279
(JNIEnv *e, jclass cl, jint fd, jlong size) {
220280
FILE_END_OF_FILE_INFO eof;
@@ -300,18 +360,24 @@ JNIEXPORT jlong JNICALL Java_io_questdb_client_std_Files_length0
300360
OPEN_EXISTING,
301361
FILE_ATTRIBUTE_NORMAL,
302362
NULL);
303-
free(wide);
304363
if (h == INVALID_HANDLE_VALUE) {
364+
/* Save the CreateFileW failure code BEFORE free(): any subsequent
365+
* API/CRT call (HeapFree can SetLastError) may overwrite the
366+
* thread's last-error value -- same pattern as open_file(). */
305367
SaveLastError();
368+
free(wide);
306369
return -1;
307370
}
371+
free(wide);
308372
LARGE_INTEGER sz;
309373
BOOL ok = GetFileSizeEx(h, &sz);
310-
CloseHandle(h);
311374
if (!ok) {
375+
/* Save before CloseHandle() can clobber the last-error value. */
312376
SaveLastError();
377+
CloseHandle(h);
313378
return -1;
314379
}
380+
CloseHandle(h);
315381
return (jlong) sz.QuadPart;
316382
}
317383

@@ -331,6 +397,20 @@ JNIEXPORT jint JNICALL Java_io_questdb_client_std_Files_lock
331397
return 0;
332398
}
333399

400+
JNIEXPORT jint JNICALL Java_io_questdb_client_cutlass_qwp_client_sf_cursor_SlotLock_release0
401+
(JNIEnv *e, jclass cl, jint fd) {
402+
OVERLAPPED ov;
403+
memset(&ov, 0, sizeof(ov));
404+
if (!UnlockFileEx(FD_TO_HANDLE(fd), 0, MAXDWORD, MAXDWORD, &ov)) {
405+
SaveLastError();
406+
return -1;
407+
}
408+
/* Unlock success confirms that the slot is reusable. Match POSIX by
409+
* closing once without making handle cleanup part of that signal. */
410+
(void) CloseHandle(FD_TO_HANDLE(fd));
411+
return 0;
412+
}
413+
334414
JNIEXPORT jint JNICALL Java_io_questdb_client_std_Files_mkdir0
335415
(JNIEnv *e, jclass cl, jlong lpszPath, jint mode) {
336416
(void) mode;
@@ -340,11 +420,13 @@ JNIEXPORT jint JNICALL Java_io_questdb_client_std_Files_mkdir0
340420
return -1;
341421
}
342422
BOOL ok = CreateDirectoryW(wide, NULL);
343-
free(wide);
344423
if (!ok) {
424+
/* Save before free() can clobber the last-error value. */
345425
SaveLastError();
426+
free(wide);
346427
return -1;
347428
}
429+
free(wide);
348430
return 0;
349431
}
350432

@@ -373,11 +455,13 @@ JNIEXPORT jboolean JNICALL Java_io_questdb_client_std_Files_remove0
373455
} else {
374456
ok = DeleteFileW(wide);
375457
}
376-
free(wide);
377458
if (!ok) {
459+
/* Save before free() can clobber the last-error value. */
378460
SaveLastError();
461+
free(wide);
379462
return JNI_FALSE;
380463
}
464+
free(wide);
381465
return JNI_TRUE;
382466
}
383467

@@ -395,12 +479,15 @@ JNIEXPORT jint JNICALL Java_io_questdb_client_std_Files_rename0
395479
return -1;
396480
}
397481
BOOL ok = MoveFileExW(oldW, newW, MOVEFILE_REPLACE_EXISTING);
398-
free(oldW);
399-
free(newW);
400482
if (!ok) {
483+
/* Save before free() can clobber the last-error value. */
401484
SaveLastError();
485+
free(oldW);
486+
free(newW);
402487
return -1;
403488
}
489+
free(oldW);
490+
free(newW);
404491
return 0;
405492
}
406493

@@ -438,24 +525,28 @@ JNIEXPORT jlong JNICALL Java_io_questdb_client_std_Files_findFirst0
438525
pattern[pathLen] = '\0';
439526

440527
wchar_t *wide = utf8_to_wide(pattern);
441-
free(pattern);
442528
if (!wide) {
529+
/* Save before free() can clobber the last-error value. */
443530
SaveLastError();
531+
free(pattern);
444532
return 0;
445533
}
534+
free(pattern);
446535

447536
qdb_find_t *find = (qdb_find_t *) malloc(sizeof(qdb_find_t));
448537
if (!find) {
449538
free(wide);
450539
return 0;
451540
}
452541
find->handle = FindFirstFileW(wide, &find->data);
453-
free(wide);
454542
if (find->handle == INVALID_HANDLE_VALUE) {
543+
/* Save before free() can clobber the last-error value. */
455544
SaveLastError();
545+
free(wide);
456546
free(find);
457547
return 0;
458548
}
549+
free(wide);
459550
find->hasEntry = 1;
460551
win_findname_to_utf8(find);
461552
return (jlong) (uintptr_t) find;
@@ -615,3 +706,16 @@ JNIEXPORT jint JNICALL Java_io_questdb_client_std_Files_msync
615706
(void) async;
616707
return 0;
617708
}
709+
710+
/* Best-effort page pin. VirtualLock is quota-bound to the process working-set
711+
* minimum; callers treat any refusal as a soft downgrade, so no
712+
* SaveLastError() -- the refusal is not surfaced as an error. */
713+
JNIEXPORT jint JNICALL Java_io_questdb_client_std_Files_mlock0
714+
(JNIEnv *e, jclass cl, jlong addr, jlong len) {
715+
return VirtualLock((LPVOID) (uintptr_t) addr, (SIZE_T) len) ? 0 : -1;
716+
}
717+
718+
JNIEXPORT jint JNICALL Java_io_questdb_client_std_Files_munlock0
719+
(JNIEnv *e, jclass cl, jlong addr, jlong len) {
720+
return VirtualUnlock((LPVOID) (uintptr_t) addr, (SIZE_T) len) ? 0 : -1;
721+
}

0 commit comments

Comments
 (0)