feat(ilp): QWiP store-and-forward client buffer

Opt-in durable buffer for the QWP WebSocket ingest client. Outgoing batches
are persisted to disk before they leave the wire; the server's cumulative
ACK trims sealed segments; on restart or transient failure, the I/O thread
silently reconnects and replays whatever is still on disk.

On-disk format is the QWP wire frame captured verbatim, wrapped in an
8-byte SF envelope (CRC32C + length) so torn tails and silent bit-rot are
caught on recovery. Filenames encode (baseSeq, lastSeq) so trim and
recovery don't have to scan sealed segments.

Auto-reconnect absorbs all transient connection failures with exponential
backoff (capped at 30s); only fatal SF storage errors (corruption, frame
larger than segment cap) propagate to the user. flush() under SF returns
once data is on disk, not on server ACK — natural backpressure when SF
total disk cap is reached makes flush() block until ACKs free space.

All file I/O goes through a new native Files layer ported from the
upstream QuestDB server repo: open/read/write/fsync/truncate/allocate/
length/lock/mkdir/exists/remove/rename/dir-iteration. Software CRC32C
implementation (Castagnoli, polynomial 0x1EDC6F41) added alongside.

Configured via connect string:

  ws::addr=...
     ;store_and_forward=on
     ;sf_dir=/var/lib/qdb/sf
     ;sf_max_bytes=67108864       (per-segment rotation; default 64 MiB)
     ;sf_max_total_bytes=4G       (hard cap → backpressure; default unlimited)
     ;sf_fsync=on                 (fsync after every append; default off)

49 new tests across 6 files: native Files (9), CRC32C (7 incl. property-
fuzz + bit-flip), SegmentLog (14), SegmentLog torture (7 incl. randomized
op-sequence fuzzer + multi-crash), SF integration (10 incl. multi-
reconnect + replay-during-replay + stress), connect-string from-config
(11). 1956 tests pass total.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: bluestreak01

core/CMakeLists.txt

@@ -53,6 +53,7 @@ set(
         src/main/c/share/cpprt_overrides.cpp
         src/main/c/share/byte_sink.cpp
         src/main/c/share/byte_sink.h
+        src/main/c/share/crc32c.c
 )
 
 # libzstd is included via a git submodule at src/main/c/share/zstd (pinned to

core/src/main/c/share/crc32c.c

@@ -0,0 +1,60 @@
+/*+*****************************************************************************
+ *     ___                  _   ____  ____
+ *    / _ \ _   _  ___  ___| |_|  _ \| __ )
+ *   | | | | | | |/ _ \/ __| __| | | |  _ \
+ *   | |_| | |_| |  __/\__ \ |_| |_| | |_) |
+ *    \__\_\\__,_|\___||___/\__|____/|____/
+ *
+ *  Copyright (c) 2014-2019 Appsicle
+ *  Copyright (c) 2019-2026 QuestDB
+ *
+ *  Licensed under the Apache License, Version 2.0 (the "License");
+ *  you may not use this file except in compliance with the License.
+ *  You may obtain a copy of the License at
+ *
+ *  http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ *
+ ******************************************************************************/
+
+#include <jni.h>
+#include <stdint.h>
+#include <stddef.h>
+
+#define CRC32C_POLY_REVERSED 0x82F63B78u
+
+static uint32_t crc32c_table[256];
+static volatile int crc32c_table_ready = 0;
+
+static void crc32c_init(void) {
+    for (int i = 0; i < 256; i++) {
+        uint32_t c = (uint32_t) i;
+        for (int j = 0; j < 8; j++) {
+            c = (c & 1u) ? (c >> 1) ^ CRC32C_POLY_REVERSED : (c >> 1);
+        }
+        crc32c_table[i] = c;
+    }
+    crc32c_table_ready = 1;
+}
+
+JNIEXPORT jint JNICALL Java_io_questdb_client_std_Crc32c_update
+        (JNIEnv *e, jclass cl, jint seed, jlong addr, jlong len) {
+    if (len <= 0) {
+        return seed;
+    }
+    if (!crc32c_table_ready) {
+        crc32c_init();
+    }
+    uint32_t crc = ~((uint32_t) seed);
+    const uint8_t *buf = (const uint8_t *) (uintptr_t) addr;
+    size_t n = (size_t) len;
+    while (n--) {
+        crc = (crc >> 8) ^ crc32c_table[(crc ^ *buf++) & 0xffu];
+    }
+    return (jint) ~crc;
+}

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

@@ -22,10 +22,249 @@
  *
  ******************************************************************************/
 
+#define _GNU_SOURCE
+
 #include <unistd.h>
+#include <sys/stat.h>
+#include <sys/file.h>
+#include <fcntl.h>
+#include <errno.h>
+#include <stdlib.h>
+#include <string.h>
+#include <dirent.h>
+#include <stdint.h>
+
 #include "files.h"
 
+#define RESTARTABLE(_expr_, _rc_) \
+    do { _rc_ = (_expr_); } while ((_rc_) == -1 && errno == EINTR)
+
 JNIEXPORT jint JNICALL Java_io_questdb_client_std_Files_close0
         (JNIEnv *e, jclass cl, jint fd) {
     return close((int) fd);
 }
+
+JNIEXPORT jint JNICALL Java_io_questdb_client_std_Files_openRO0
+        (JNIEnv *e, jclass cl, jlong lpszName) {
+    int fd;
+    RESTARTABLE(open((const char *) (uintptr_t) lpszName, O_RDONLY), fd);
+    return (jint) fd;
+}
+
+JNIEXPORT jint JNICALL Java_io_questdb_client_std_Files_openRW0
+        (JNIEnv *e, jclass cl, jlong lpszName) {
+    int fd;
+    RESTARTABLE(open((const char *) (uintptr_t) lpszName, O_CREAT | 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;
+    RESTARTABLE(open((const char *) (uintptr_t) lpszName, O_CREAT | O_WRONLY | O_APPEND, 0644), fd);
+    return (jint) fd;
+}
+
+JNIEXPORT jint JNICALL Java_io_questdb_client_std_Files_openCleanRW0
+        (JNIEnv *e, jclass cl, jlong lpszName, jlong size) {
+    int fd;
+    RESTARTABLE(open((const char *) (uintptr_t) lpszName, O_CREAT | O_TRUNC | O_RDWR, 0644), fd);
+    if (fd < 0) {
+        return -1;
+    }
+    if (size > 0) {
+        int rc;
+        RESTARTABLE(ftruncate(fd, (off_t) size), rc);
+        if (rc != 0) {
+            int saved = errno;
+            close(fd);
+            errno = saved;
+            return -1;
+        }
+    }
+    return (jint) fd;
+}
+
+JNIEXPORT jlong JNICALL Java_io_questdb_client_std_Files_read
+        (JNIEnv *e, jclass cl, jint fd, jlong addr, jlong len, jlong offset) {
+    ssize_t res;
+    RESTARTABLE(pread((int) fd, (void *) (uintptr_t) addr, (size_t) len, (off_t) offset), res);
+    return (jlong) res;
+}
+
+JNIEXPORT jlong JNICALL Java_io_questdb_client_std_Files_write
+        (JNIEnv *e, jclass cl, jint fd, jlong addr, jlong len, jlong offset) {
+    ssize_t res;
+    RESTARTABLE(pwrite((int) fd, (const void *) (uintptr_t) addr, (size_t) len, (off_t) offset), res);
+    return (jlong) res;
+}
+
+JNIEXPORT jlong JNICALL Java_io_questdb_client_std_Files_append
+        (JNIEnv *e, jclass cl, jint fd, jlong addr, jlong len) {
+    ssize_t res;
+    RESTARTABLE(write((int) fd, (const void *) (uintptr_t) addr, (size_t) len), res);
+    return (jlong) res;
+}
+
+JNIEXPORT jint JNICALL Java_io_questdb_client_std_Files_fsync
+        (JNIEnv *e, jclass cl, jint fd) {
+    int res;
+    RESTARTABLE(fsync((int) fd), res);
+    return res;
+}
+
+JNIEXPORT jboolean JNICALL Java_io_questdb_client_std_Files_truncate
+        (JNIEnv *e, jclass cl, jint fd, jlong size) {
+    int res;
+    RESTARTABLE(ftruncate((int) fd, (off_t) size), res);
+    return res == 0 ? JNI_TRUE : JNI_FALSE;
+}
+
+JNIEXPORT jboolean JNICALL Java_io_questdb_client_std_Files_allocate
+        (JNIEnv *e, jclass cl, jint fd, jlong size) {
+#if defined(__linux__)
+    int res = posix_fallocate((int) fd, 0, (off_t) size);
+    if (res == 0) {
+        return JNI_TRUE;
+    }
+    if (res != EINVAL && res != EOPNOTSUPP) {
+        errno = res;
+        return JNI_FALSE;
+    }
+    /* fall through to ftruncate */
+#elif defined(__APPLE__)
+    fstore_t fst;
+    fst.fst_flags = F_ALLOCATECONTIG | F_ALLOCATEALL;
+    fst.fst_posmode = F_PEOFPOSMODE;
+    fst.fst_offset = 0;
+    fst.fst_length = (off_t) size;
+    fst.fst_bytesalloc = 0;
+    if (fcntl((int) fd, F_PREALLOCATE, &fst) == -1) {
+        fst.fst_flags = F_ALLOCATEALL;
+        (void) fcntl((int) fd, F_PREALLOCATE, &fst);
+        /* if F_PREALLOCATE fails we still try ftruncate to set logical size */
+    }
+#endif
+    int res2;
+    RESTARTABLE(ftruncate((int) fd, (off_t) size), res2);
+    return res2 == 0 ? JNI_TRUE : JNI_FALSE;
+}
+
+JNIEXPORT jlong JNICALL Java_io_questdb_client_std_Files_length
+        (JNIEnv *e, jclass cl, jint fd) {
+    struct stat st;
+    if (fstat((int) fd, &st) != 0) {
+        return -1;
+    }
+    return (jlong) st.st_size;
+}
+
+JNIEXPORT jlong JNICALL Java_io_questdb_client_std_Files_length0
+        (JNIEnv *e, jclass cl, jlong lpszName) {
+    struct stat st;
+    if (stat((const char *) (uintptr_t) lpszName, &st) != 0) {
+        return -1;
+    }
+    return (jlong) st.st_size;
+}
+
+JNIEXPORT jint JNICALL Java_io_questdb_client_std_Files_lock
+        (JNIEnv *e, jclass cl, jint fd) {
+    return flock((int) fd, LOCK_EX | LOCK_NB);
+}
+
+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);
+}
+
+JNIEXPORT jboolean JNICALL Java_io_questdb_client_std_Files_exists0
+        (JNIEnv *e, jclass cl, jlong lpszPath) {
+    return access((const char *) (uintptr_t) lpszPath, F_OK) == 0 ? JNI_TRUE : JNI_FALSE;
+}
+
+JNIEXPORT jboolean JNICALL Java_io_questdb_client_std_Files_remove0
+        (JNIEnv *e, jclass cl, jlong lpszPath) {
+    return remove((const char *) (uintptr_t) lpszPath) == 0 ? JNI_TRUE : JNI_FALSE;
+}
+
+JNIEXPORT jint JNICALL Java_io_questdb_client_std_Files_rename0
+        (JNIEnv *e, jclass cl, jlong lpszOld, jlong lpszNew) {
+    return rename((const char *) (uintptr_t) lpszOld, (const char *) (uintptr_t) lpszNew);
+}
+
+typedef struct {
+    DIR *dir;
+    struct dirent *entry;
+} qdb_find_t;
+
+JNIEXPORT jlong JNICALL Java_io_questdb_client_std_Files_findFirst0
+        (JNIEnv *e, jclass cl, jlong lpszName) {
+    DIR *dir = opendir((const char *) (uintptr_t) lpszName);
+    if (!dir) {
+        return 0;
+    }
+    qdb_find_t *find = (qdb_find_t *) malloc(sizeof(qdb_find_t));
+    if (!find) {
+        closedir(dir);
+        return 0;
+    }
+    find->dir = dir;
+    errno = 0;
+    find->entry = readdir(dir);
+    if (!find->entry) {
+        int saved = errno;
+        closedir(dir);
+        free(find);
+        errno = saved;
+        return 0;
+    }
+    return (jlong) (uintptr_t) find;
+}
+
+JNIEXPORT jint JNICALL Java_io_questdb_client_std_Files_findNext
+        (JNIEnv *e, jclass cl, jlong findPtr) {
+    qdb_find_t *find = (qdb_find_t *) (uintptr_t) findPtr;
+    if (!find) {
+        return -1;
+    }
+    errno = 0;
+    find->entry = readdir(find->dir);
+    if (find->entry) {
+        return 1;
+    }
+    return errno == 0 ? 0 : -1;
+}
+
+JNIEXPORT jlong JNICALL Java_io_questdb_client_std_Files_findName
+        (JNIEnv *e, jclass cl, jlong findPtr) {
+    qdb_find_t *find = (qdb_find_t *) (uintptr_t) findPtr;
+    if (!find || !find->entry) {
+        return 0;
+    }
+    return (jlong) (uintptr_t) find->entry->d_name;
+}
+
+JNIEXPORT jint JNICALL Java_io_questdb_client_std_Files_findType
+        (JNIEnv *e, jclass cl, jlong findPtr) {
+    qdb_find_t *find = (qdb_find_t *) (uintptr_t) findPtr;
+    if (!find || !find->entry) {
+        return 0;
+    }
+    return (jint) find->entry->d_type;
+}
+
+JNIEXPORT void JNICALL Java_io_questdb_client_std_Files_findClose
+        (JNIEnv *e, jclass cl, jlong findPtr) {
+    qdb_find_t *find = (qdb_find_t *) (uintptr_t) findPtr;
+    if (find) {
+        closedir(find->dir);
+        free(find);
+    }
+}
+
+JNIEXPORT jlong JNICALL Java_io_questdb_client_std_Files_getPageSize0
+        (JNIEnv *e, jclass cl) {
+    long sz = sysconf(_SC_PAGESIZE);
+    return (jlong) (sz > 0 ? sz : 4096);
+}