feat(ilp): cursor SF engine primitives (mmap segments, ring, manager)

Lays the foundation for a lock-free, mmap-backed alternative to the
current SegmentLog + WebSocketSendQueue + processingLock design. Today
~85% of the user thread's flush() time is spent parked in
__psynch_cvwait waiting for the I/O thread to signal completion (see
QwpIngressLatencyBenchmark async-profiler flamegraph). The cursor design
moves SF.append onto the user thread, making the cross-thread wait
unnecessary -- the user-thread append microbench now reports p50=42ns
vs ~38us in the legacy SF path on the same hardware.

What lands:

  * mmap/munmap/msync ported from QuestDB OSS into client/std/Files
    (both POSIX and Win32). Native rebuild required per-platform; the
    darwin-aarch64 dev lib is the only one rebuilt locally.

  * MmapSegment: one mmap'd file, format-compatible with the legacy
    SegmentLog (same SF01 magic, 24-byte header, [crc | u32 len |
    payload] frame layout). Single-producer cursor (appendCursor plain
    field, publishedCursor volatile). tryAppend is pure memory + CRC.
    openExisting + scanFrames recover from torn tails.

  * SegmentRing: chain of MmapSegments with hot-spare swap and
    ACK-driven trim. Four cursors, all single-writer (no CAS). Rotation
    rebases the spare's baseSeq at promotion time to avoid the
    precompute race.

  * SegmentManager: JVM-wide background thread that pre-creates spares
    and trims fully-acked segments. Moves the open + truncate + fsync +
    rename + unlink quartet (45k samples / 100k I/O thread samples in
    the legacy flamegraph) off the hot path.

  * CursorSendEngine: facade bundling ring + manager with the API a
    future WebSocketSendQueue rewrite will consume.

  * sf_engine=legacy|cursor config option in LineSenderBuilder. Default
    legacy. Selecting cursor at build time fails fast with a clear "not
    yet wired" message -- the WebSocketSendQueue integration that
    actually consumes CursorSendEngine is the next PR.

  * CursorEngineAppendLatencyBenchmark: standalone microbench for the
    user-thread append path (the floor a wired cursor engine would
    inherit).

20 new tests across the cursor/ package, all green. FilesTest gains a
mmap roundtrip test.

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

Author: bluestreak01

.gitignore

@@ -21,6 +21,7 @@ core/questdb/client/bin-local
 core/cmake-build-debug
 core/cmake-build-debug-coverage
 core/cmake-build-release
+core/build_native
 core/CMakeCache.txt
 **/.project
 **/.settings

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

@@ -27,6 +27,7 @@
 #include <unistd.h>
 #include <sys/stat.h>
 #include <sys/file.h>
+#include <sys/mman.h>
 #include <fcntl.h>
 #include <errno.h>
 #include <stdlib.h>
@@ -36,6 +37,12 @@
 
 #include "files.h"
 
+/* Mirror of io.questdb.client.std.Files.MAP_RO / MAP_RW. Hard-coded rather
+ * than #include'd from a javah-generated header because this file does not
+ * pull in any generated symbols (the rest of the file works the same way). */
+#define QDB_MAP_RO 1
+#define QDB_MAP_RW 2
+
 #define RESTARTABLE(_expr_, _rc_) \
     do { _rc_ = (_expr_); } while ((_rc_) == -1 && errno == EINTR)
 
@@ -268,3 +275,26 @@ JNIEXPORT jlong JNICALL Java_io_questdb_client_std_Files_getPageSize0
     long sz = sysconf(_SC_PAGESIZE);
     return (jlong) (sz > 0 ? sz : 4096);
 }
+
+JNIEXPORT jlong JNICALL Java_io_questdb_client_std_Files_mmap0
+        (JNIEnv *e, jclass cl, jint fd, jlong len, jlong offset, jint flags, jlong baseAddress) {
+    int prot = 0;
+    if (flags == QDB_MAP_RO) {
+        prot = PROT_READ;
+    } else if (flags == QDB_MAP_RW) {
+        prot = PROT_READ | PROT_WRITE;
+    }
+    void *addr = mmap((void *) (uintptr_t) baseAddress, (size_t) len, prot, MAP_SHARED, (int) fd, (off_t) offset);
+    /* MAP_FAILED is (void *) -1; cast to jlong gives -1 sentinel matching FAILED_MMAP_ADDRESS. */
+    return (jlong) (intptr_t) addr;
+}
+
+JNIEXPORT jint JNICALL Java_io_questdb_client_std_Files_munmap0
+        (JNIEnv *e, jclass cl, jlong address, jlong len) {
+    return munmap((void *) (uintptr_t) address, (size_t) len);
+}
+
+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);
+}

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

@@ -468,3 +468,95 @@ JNIEXPORT jlong JNICALL Java_io_questdb_client_std_Files_getPageSize0
     GetSystemInfo(&si);
     return (jlong) si.dwAllocationGranularity;
 }
+
+/* Mirror of io.questdb.client.std.Files.MAP_RO / MAP_RW. */
+#define QDB_MAP_RO 1
+#define QDB_MAP_RW 2
+
+JNIEXPORT jlong JNICALL Java_io_questdb_client_std_Files_mmap0
+        (JNIEnv *e, jclass cl, jint fd, jlong len, jlong offset, jint flags, jlong baseAddress) {
+    if (len == 0) {
+        /* Win32 MapViewOfFileEx interprets dwNumberOfBytesToMap == 0 as
+         * "map to end of mapping". Reject explicitly so the wrapper has
+         * POSIX-compatible semantics (POSIX mmap with len==0 returns
+         * EINVAL). */
+        SetLastError(ERROR_INVALID_PARAMETER);
+        SaveLastError();
+        return -1;
+    }
+
+    jlong maxsize = offset + len;
+    DWORD flProtect;
+    DWORD dwDesiredAccess;
+    if (flags == QDB_MAP_RW) {
+        flProtect = PAGE_READWRITE;
+        dwDesiredAccess = FILE_MAP_WRITE;
+    } else {
+        flProtect = PAGE_READONLY;
+        dwDesiredAccess = FILE_MAP_READ;
+    }
+
+    HANDLE hMapping = CreateFileMapping(
+            FD_TO_HANDLE(fd),
+            NULL,
+            flProtect | SEC_RESERVE,
+            (DWORD) (maxsize >> 32),
+            (DWORD) maxsize,
+            NULL);
+    if (hMapping == NULL) {
+        SaveLastError();
+        return -1;
+    }
+
+    LPCVOID address = MapViewOfFileEx(
+            hMapping,
+            dwDesiredAccess,
+            (DWORD) (offset >> 32),
+            (DWORD) offset,
+            (SIZE_T) len,
+            (LPVOID) (uintptr_t) baseAddress);
+
+    SaveLastError();
+
+    /* The mapping handle can be closed immediately — the view holds its own
+     * reference and the file mapping persists until the last view is unmapped. */
+    if (CloseHandle(hMapping) == 0) {
+        SaveLastError();
+        if (address != NULL) {
+            UnmapViewOfFile(address);
+        }
+        return -1;
+    }
+
+    if (address == NULL) {
+        return -1;
+    }
+    return (jlong) (uintptr_t) address;
+}
+
+JNIEXPORT jint JNICALL Java_io_questdb_client_std_Files_munmap0
+        (JNIEnv *e, jclass cl, jlong address, jlong len) {
+    if (UnmapViewOfFile((LPCVOID) (uintptr_t) address) == 0) {
+        SaveLastError();
+        return -1;
+    }
+    return 0;
+}
+
+JNIEXPORT jint JNICALL Java_io_questdb_client_std_Files_msync
+        (JNIEnv *e, jclass cl, jlong addr, jlong len, jboolean async) {
+    /* FlushViewOfFile schedules a write, blocking until the file system
+     * driver has accepted the write into its cache. For "fully durable"
+     * (POSIX MS_SYNC equivalent) we need a follow-up FlushFileBuffers,
+     * but that needs the file handle which we no longer hold here.
+     * MS_ASYNC maps cleanly: don't wait for further confirmation. */
+    if (FlushViewOfFile((LPCVOID) (uintptr_t) addr, (SIZE_T) len) == 0) {
+        SaveLastError();
+        return -1;
+    }
+    /* We deliberately do NOT call FlushFileBuffers in the async case;
+     * sync callers wanting the strongest durability should fsync the fd
+     * separately via Files.fsync. */
+    (void) async;
+    return 0;
+}

core/src/main/java/io/questdb/client/Sender.java

@@ -632,6 +632,12 @@ public int getTimeout() {
         private long sfMaxTotalBytes = PARAMETER_NOT_SET_EXPLICITLY;
         private boolean sfFsync;
         private boolean sfFsyncOnFlush;
+        // SF storage engine: "legacy" = SegmentLog + WebSocketSendQueue (today's
+        // default). "cursor" = mmap-backed SegmentRing + lock-free cursor design
+        // (in-progress; not yet wired into the Sender — selecting it at build
+        // time fails fast). null = parameter not explicitly set (defaults to
+        // "legacy").
+        private String sfEngine;
         private boolean tlsEnabled;
         private TlsValidationMode tlsValidationMode;
         private char[] trustStorePassword;
@@ -934,6 +940,19 @@ public Sender build() {
                     );
                 }
 
+                // Engine selection (legacy default). The cursor engine is
+                // implemented (see io.questdb.client.cutlass.qwp.client.sf.cursor)
+                // but not yet plumbed through QwpWebSocketSender — fail fast
+                // here instead of silently falling back to legacy.
+                if ("cursor".equals(sfEngine)) {
+                    throw new LineSenderException(
+                            "sf_engine=cursor is not yet wired into the Sender — the engine "
+                                    + "primitives (MmapSegment / SegmentRing / SegmentManager / "
+                                    + "CursorSendEngine) are in place but the WebSocketSendQueue "
+                                    + "rewrite that consumes them is the next PR. Track the "
+                                    + "follow-up issue and use sf_engine=legacy in the meantime.");
+                }
+
                 SegmentLog segmentLog = null;
                 if (storeAndForward) {
                     if (sfDir == null) {
@@ -1689,6 +1708,31 @@ public LineSenderBuilder storeAndForwardFsyncOnFlush(boolean enabled) {
             return this;
         }
 
+        /**
+         * Selects the SF storage engine. Allowed values:
+         * <ul>
+         *   <li>{@code "legacy"} — pwrite-based {@code SegmentLog} routed
+         *       through {@code WebSocketSendQueue}. Today's default.</li>
+         *   <li>{@code "cursor"} — mmap-backed {@code SegmentRing} with a
+         *       background segment manager and a lock-free user-thread
+         *       append path. Substantially lower per-flush latency. NOT YET
+         *       WIRED into {@code QwpWebSocketSender}; selecting it at build
+         *       time throws {@link LineSenderException} so users can't
+         *       silently fall back to legacy. Tracking issue / future PR.</li>
+         * </ul>
+         */
+        public LineSenderBuilder storeAndForwardEngine(String engine) {
+            if (protocol != PARAMETER_NOT_SET_EXPLICITLY && protocol != PROTOCOL_WEBSOCKET) {
+                throw new LineSenderException("sf_engine is only supported for WebSocket transport");
+            }
+            if (!"legacy".equals(engine) && !"cursor".equals(engine)) {
+                throw new LineSenderException("invalid sf_engine [value=").put(engine)
+                        .put(", allowed-values=[legacy, cursor]]");
+            }
+            this.sfEngine = engine;
+            return this;
+        }
+
         /**
          * Configures the maximum time the Sender will spend retrying upon receiving a recoverable error from the server.
          * <br>
@@ -2167,6 +2211,12 @@ private LineSenderBuilder fromConfig(CharSequence configurationString) {
                     } else {
                         throw new LineSenderException("invalid sf_fsync_on_flush [value=").put(sink).put(", allowed-values=[on, off]]");
                     }
+                } else if (Chars.equals("sf_engine", sink)) {
+                    if (protocol != PROTOCOL_WEBSOCKET) {
+                        throw new LineSenderException("sf_engine is only supported for WebSocket transport");
+                    }
+                    pos = getValue(configurationString, pos, sink, "sf_engine");
+                    storeAndForwardEngine(sink.toString());
                 } else if (Chars.equals("max_datagram_size", sink)) {
                     pos = getValue(configurationString, pos, sink, "max_datagram_size");
                     int mds = parseIntValue(sink, "max_datagram_size");