fix(qwp): make SF barrier retries genuine re-persists (C-1 unsound latch clear)

After a failed writeback the kernel marks the dirtied pages clean and
reports the error once per open file description (errseq_t; fsyncgate),
so re-running msync+fsync over the same range returns a vacuous 0:
durableCursor advanced, the PERIODIC rotation gate opened, and the
round-3 unlatch (958a36d) resumed the producer -- all without the data
being durable. A clean non-durable page is also evictable, so reclaim
could silently replace the only good copy with stale disk content.

- MmapSegment.syncPublished now brackets the barrier: best-effort mlock
  pin over the page-aligned [durableCursor, published) range, then
  msync+fsync; the failure arm re-dirties the pinned range (one
  same-value store per page -- kernel dirty tracking is value-blind)
  before the pin is released. The pin closes the clean-page reclaim race
  by construction; the re-dirty guarantees the next barrier performs
  real writeback and reports real errors, keeping the manager's
  clearDurabilityFailure() honest.
- mlock/munlock natives added (POSIX mlock, Windows VirtualLock) and
  exposed through Files with an UnsatisfiedLinkError guard plus
  FilesFacade defaults. Refusal (RLIMIT_MEMLOCK, missing capability) is
  a soft downgrade: one deduped WARN, barrier outcome and latch
  semantics unaffected.
- 4 regression tests: bracket ordering on success, re-dirty-before-unpin
  on failure, an fsyncgate facade model (first fsync fails, later
  msync/fsync return vacuous 0 without delegating) proving the latch
  only clears over re-dirtied pages, and mlock-refusal degrade. The
  drop-the-redirty mutant was applied and killed (2 failures) before
  restoring.

Native libraries are built from source, so the new symbols ship with
every build; the UnsatisfiedLinkError guard remains as defense-in-depth
for environments running a stale prebuilt library.

Author: bluestreak01

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

@@ -388,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/windows/files.c

@@ -689,3 +689,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;
+}

core/src/main/java/io/questdb/client/cutlass/qwp/client/sf/cursor/MmapSegment.java

@@ -31,9 +31,12 @@
 import io.questdb.client.std.Os;
 import io.questdb.client.std.QuietCloseable;
 import io.questdb.client.std.Unsafe;
+import org.jetbrains.annotations.TestOnly;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.util.concurrent.atomic.AtomicBoolean;
+
 /**
  * One mmap-backed SF segment file. The user thread (the single producer)
  * appends frames into the mapping; the I/O thread (the single consumer) reads
@@ -68,6 +71,10 @@ public final class MmapSegment implements QuietCloseable {
     public static final byte MANIFEST_REQUIRED_FLAG = 1;
     public static final byte VERSION = 1;
     private static final Logger LOG = LoggerFactory.getLogger(MmapSegment.class);
+    // Deduplicates the process-wide "mlock refused" warning: the refusal is a
+    // soft downgrade (see syncPublished) and must not spam the log once per
+    // barrier when RLIMIT_MEMLOCK or the platform says no.
+    private static final AtomicBoolean MLOCK_REFUSAL_WARNED = new AtomicBoolean();
     private static final int RECOVERY_BUFFER_SIZE = 64 * 1024;
 
     private final FilesFacade filesFacade;
@@ -103,6 +110,9 @@ public final class MmapSegment implements QuietCloseable {
     // because the consumer must see writes in publication order — once the
     // producer bumps publishedCursor, every byte before it is fully written.
     private volatile long publishedCursor;
+    // Number of failure-arm re-dirty passes performed by syncPublished.
+    // Cold-path only; volatile so tests observing from another thread see it.
+    private volatile long redirtyPasses;
     // Monotonic in-memory link to the segment that immediately follows this
     // one. SegmentRing publishes it before promoting the successor to active;
     // close deliberately retains it so a cursor can advance after head trim.
@@ -486,6 +496,21 @@ public void msync() {
      * captures {@link #publishedCursor}. A concurrent producer may publish
      * more bytes while the barrier runs; those bytes remain outside the
      * returned durable boundary until a later call.
+     * <p>
+     * The not-yet-durable range is pinned with a best-effort {@code mlock}
+     * for the duration of the barrier and, on failure, re-dirtied before the
+     * pin is released. Rationale (fsyncgate): after a failed writeback the
+     * kernel marks the affected pages clean and reports the error once per
+     * open file description, so a naive retry would msync+fsync clean pages
+     * and return a vacuous 0 without persisting anything -- and an unpinned
+     * clean page can be reclaimed and later re-faulted from stale disk
+     * content, silently replacing the only good copy of the frames. The pin
+     * guarantees the failure arm re-dirties the genuine in-memory bytes; the
+     * re-dirty guarantees the next barrier performs real writeback and
+     * reports real errors, so the manager's unlatch-on-success stays honest.
+     * A refused mlock (RLIMIT_MEMLOCK, missing capability, stale native
+     * library) never affects the barrier outcome; it only widens the
+     * microseconds-scale window between the failed syscall and the re-dirty.
      *
      * @return the captured byte offset covered by the successful barrier
      */
@@ -498,16 +523,55 @@ public long syncPublished() {
         if (published <= durableCursor) {
             return durableCursor;
         }
-        if (filesFacade.msync(mmapAddress, published, false) != 0) {
-            throw new MmapSegmentException("could not sync segment data " + path);
+        long lockOffset = durableCursor & -Files.PAGE_SIZE;
+        long lockAddr = mmapAddress + lockOffset;
+        long lockLen = published - lockOffset;
+        boolean locked = filesFacade.mlock(lockAddr, lockLen) == 0;
+        if (!locked && MLOCK_REFUSAL_WARNED.compareAndSet(false, true)) {
+            LOG.warn("mlock refused for SF barrier range in {} (degrading to re-dirty-only retry protection); "
+                    + "raise RLIMIT_MEMLOCK or grant CAP_IPC_LOCK to close the post-failure reclaim window", path);
         }
-        // FlushViewOfFile alone is not power-loss durable on Windows. Keep the
-        // fd barrier on every platform so this method has one portable contract.
-        if (filesFacade.fsync(fd) != 0) {
-            throw new MmapSegmentException("could not sync segment file " + path);
+        boolean durable = false;
+        try {
+            if (filesFacade.msync(mmapAddress, published, false) != 0) {
+                throw new MmapSegmentException("could not sync segment data " + path);
+            }
+            // FlushViewOfFile alone is not power-loss durable on Windows. Keep the
+            // fd barrier on every platform so this method has one portable contract.
+            if (filesFacade.fsync(fd) != 0) {
+                throw new MmapSegmentException("could not sync segment file " + path);
+            }
+            durable = true;
+            durableCursor = published;
+            return published;
+        } finally {
+            if (!durable) {
+                redirtyRange(lockAddr, mmapAddress + published);
+            }
+            if (locked) {
+                filesFacade.munlock(lockAddr, lockLen);
+            }
+        }
+    }
+
+    /**
+     * Marks every page overlapping {@code [fromAddr, endAddr)} dirty again by
+     * re-storing one byte per page. Every touched offset lies inside the
+     * published prefix (or the immutable header magic after aligning down),
+     * so the same-value store races with nothing; kernel dirty tracking is
+     * value-blind, so the store is a real dirtying event that forces the next
+     * writeback to re-submit the whole page.
+     */
+    private void redirtyRange(long fromAddr, long endAddr) {
+        for (long addr = fromAddr; addr < endAddr; addr += Files.PAGE_SIZE) {
+            Unsafe.getUnsafe().putByte(addr, Unsafe.getUnsafe().getByte(addr));
         }
-        durableCursor = published;
-        return published;
+        redirtyPasses++;
+    }
+
+    @TestOnly
+    public long redirtyPassesForTest() {
+        return redirtyPasses;
     }
 
     /**

core/src/main/java/io/questdb/client/cutlass/qwp/client/sf/cursor/SegmentManager.java

@@ -1224,9 +1224,12 @@ private void servicePeriodicSync(RingEntry e, long now) {
                 syncScratch.getQuick(i).syncPublished();
             }
             e.ring.clearSyncRequestIfActiveDurable();
-            // The pass above covered every live segment's published range, so
-            // any earlier barrier failure has been remedied — unlatch it so a
-            // transient disk fault doesn't permanently brick the producer.
+            // The pass above covered every live segment's published range, and
+            // a failed barrier re-dirties its range under an mlock pin (see
+            // MmapSegment.syncPublished), so a success here is a genuine
+            // re-persist -- not a vacuous retry over pages a failed writeback
+            // marked clean (fsyncgate). Unlatch so a transient disk fault
+            // doesn't permanently brick the producer.
             e.ring.clearDurabilityFailure();
             e.nextDataSyncNanos = now + e.syncIntervalNanos;
             if (e.syncFailureLogged) {

core/src/main/java/io/questdb/client/std/Files.java

@@ -573,6 +573,53 @@ public static void munmap(long address, long len, int memoryTag) {
      */
     public static native int msync(long addr, long len, boolean async);
 
+    // Guards the optional mlock/munlock natives: a packaged native library
+    // built before these symbols existed throws UnsatisfiedLinkError on first
+    // use. Flip to unavailable and report refusal (-1) from then on, so
+    // callers degrade to their unpinned tier instead of crashing.
+    private static volatile boolean mlockLinked = true;
+
+    /**
+     * Best-effort page pin over {@code [addr, addr+len)} of an mmap'd region.
+     * Whole pages containing any part of the range are locked; {@code addr}
+     * should be page-aligned for portability. Returns 0 on success, -1 when
+     * the platform refuses (RLIMIT_MEMLOCK, missing privilege) or the loaded
+     * native library predates the symbol. Callers must treat refusal as a
+     * soft downgrade, never an error.
+     */
+    public static int mlock(long addr, long len) {
+        if (!mlockLinked) {
+            return -1;
+        }
+        try {
+            return mlock0(addr, len);
+        } catch (UnsatisfiedLinkError err) {
+            mlockLinked = false;
+            return -1;
+        }
+    }
+
+    /**
+     * Releases a pin taken by {@link #mlock(long, long)}. Best-effort: a
+     * refusal is ignorable ({@code munmap} implicitly drops any remaining
+     * locks on the range).
+     */
+    public static int munlock(long addr, long len) {
+        if (!mlockLinked) {
+            return -1;
+        }
+        try {
+            return munlock0(addr, len);
+        } catch (UnsatisfiedLinkError err) {
+            mlockLinked = false;
+            return -1;
+        }
+    }
+
+    private static native int mlock0(long addr, long len);
+
+    private static native int munlock0(long addr, long len);
+
     /**
      * Returns a native pointer to the current entry's null-terminated name
      * (UTF-8). Pointer is valid only until the next {@link #findNext(long)}

core/src/main/java/io/questdb/client/std/FilesFacade.java

@@ -108,6 +108,16 @@ default int fsyncDir(String dir) {
 
     int mkdir(String path, int mode);
 
+    /**
+     * Best-effort page pin over {@code [addr, addr+len)} of an mmap'd region.
+     * Returns 0 when the range is locked, non-zero when the platform refuses
+     * (RLIMIT_MEMLOCK, missing privilege, or a native library without the
+     * symbol). Callers must treat refusal as a soft downgrade, never an error.
+     */
+    default int mlock(long addr, long len) {
+        return Files.mlock(addr, len);
+    }
+
     default long mmap(int fd, long len, long offset, int flags, int memoryTag) {
         return Files.mmap(fd, len, offset, flags, memoryTag);
     }
@@ -116,6 +126,14 @@ default int msync(long addr, long len, boolean async) {
         return Files.msync(addr, len, async);
     }
 
+    /**
+     * Releases a pin taken by {@link #mlock(long, long)}. Best-effort;
+     * refusals are ignorable ({@code munmap} implicitly unlocks).
+     */
+    default int munlock(long addr, long len) {
+        return Files.munlock(addr, len);
+    }
+
     default void munmap(long address, long len, int memoryTag) {
         Files.munmap(address, len, memoryTag);
     }