fix(client): fail closed in SF recovery and add crash-safe boundary manifest

SF crash recovery previously failed OPEN on operational errors:
- findFirst() < 0 (permission/transient enumeration error) was treated as
  an empty directory, so the engine started fresh and openCleanRW(O_TRUNC)
  truncated the existing durable log (sf-initial.sfa) and deleted the ack
  watermark;
- findNext() < 0 silently accepted an incomplete directory listing;
- per-file openRW/mmap failures (EMFILE, ENOMEM, transient I/O) on valid
  segments were swallowed by the same catch as data corruption and skipped;
- the FSN contiguity check only compares adjacent pairs, so a skipped
  LEADING segment silently dropped unacked rows (ackedFsn seeded past
  them) and a skipped TRAILING segment silently re-issued its FSNs to new
  payloads (overlapping FSNs on disk).

Recovery now fails closed and proves slot integrity before mutating:

- Enumeration errors (findFirst/findNext < 0) abort startup.
- Operational open/mmap failures on recognized segments abort startup;
  only positively-identified corruption (new MmapSegmentCorruptionException:
  bad magic, sub-header size, negative baseSeq, unreadable header page) is
  skippable, and quarantine renames (<name>.corrupt) are deferred until the
  surviving chain validates, so a failed recovery mutates nothing.
  Unsupported-version segments stay fatal (they belong to another client
  build; renaming them would strand that build's frames).
- sf-manifest.bin: dual-slot, CRC-protected, generation-versioned boundary
  record (headBase/activeBase) written on rotation and ack-driven trim.
  Recovery validates the chain against the committed boundaries, which
  catches missing leading/trailing segments that contiguity alone cannot.
  Segments carry a manifest-required header flag so a deleted manifest next
  to migrated segments fails closed. Boundary updates are monotonic
  (clamped) and serialized with trim under the ring monitor; the trim path
  recomputes the successor under that monitor so a concurrent rotation can
  never make the head leapfrog a still-unacked sealed segment.
- Segment files are created with O_EXCL (new Files.openRWExclusive natives)
  instead of O_TRUNC, so no code path can truncate an existing log.
- Crash-window protocol, verified state by state: fresh start orders
  initial-segment durability -> manifest create -> flag stamp; rotation
  syncs the promoted spare's rebased header before the manifest references
  it and commits the manifest before any ring mutation; clean-drain close
  durably collapses boundaries to head==active before unlinking and removes
  the manifest last. Every window either recovers (equivalent empties from
  a fresh-start kill, an empty active at the chain end after a rotation
  crash, record-less manifest creation debris, drain-window survivors,
  stale below-head files) or fails closed with intact evidence. The one
  deliberate mutation on a failing path is quarantining a provably
  record-less manifest (it contains no boundaries by construction; flagged
  segments still fail closed afterwards).

Regression tests (SegmentRecoveryIntegrityTest, 24 cases) cover the full
required matrix: findFirst=-1, partial findNext=-1, openRW=-1, mmap
failure -- each asserting byte-for-byte slot preservation; the engine-level
O_TRUNC reproduction; leading/trailing/interior boundary loss; corrupt
stray quarantine with sibling recovery; deferred-quarantine no-mutation on
failure; flagged-segments-without-manifest; manifest creation-crash debris
self-heal (zero-byte and record-less); drain-crash spare survivor;
corrupt-active-with-empty-stand-in fail-closed; dual-slot generation
selection and torn-record fallback.

Independently reviewed for crash-consistency holes, lock ordering
(manager lock -> ring monitor -> manifest monitor, acyclic), unacked-frame
loss, and fd/mmap leaks; both review blockers (creation-debris brick,
drain-window spare brick) are fixed and regression-tested. Windows native
openRWExclusive0 uses CREATE_NEW; CI rebuilds platform binaries from the
.c change.

Author: bluestreak01

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;

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

@@ -103,6 +103,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,

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

@@ -286,9 +286,11 @@ private CursorSendEngine(String sfDir, long segmentSizeBytes, SegmentManager man
             // session before deciding to start fresh. Without this the engine
             // would create a new sf-initial.sfa at baseSeq=0, overlapping FSNs
             // already on disk and corrupting ACK translation, trim, and replay.
-            SegmentRing recovered = memoryMode ? null
-                    : SegmentRing.openExisting(sfDir, segmentSizeBytes);
-            this.wasRecoveredFromDisk = recovered != null;
+            SegmentRing.Recovery recovery = memoryMode ? null
+                    : SegmentRing.recover(filesFacade, sfDir, segmentSizeBytes);
+            SegmentRing recovered = recovery == null ? null : recovery.ring();
+            this.wasRecoveredFromDisk = recovery != null
+                    && recovery.status() == SegmentRing.RecoveryStatus.RECOVERED;
             if (recovered != null) {
                 ringInProgress = recovered;
                 // Seed ackedFsn to one below the lowest segment's baseSeq.
@@ -394,18 +396,49 @@ private CursorSendEngine(String sfDir, long segmentSizeBytes, SegmentManager man
                 }
                 MmapSegment initial;
                 String initialPath = null;
+                SfManifest initialManifest = null;
                 if (memoryMode) {
                     initial = MmapSegment.createInMemory(0L, segmentSizeBytes);
                 } else {
+                    // Created WITHOUT the manifest-required flag: the flag is
+                    // a durable promise that sf-manifest.bin exists, so it
+                    // must only be stamped after the manifest itself is on
+                    // disk. Stamping it at create time would leave a crash
+                    // window (initial durable, manifest not yet created)
+                    // whose recovery hard-fails with "segment requires
+                    // missing manifest" even though nothing was lost.
                     initialPath = sfDir + "/sf-initial.sfa";
-                    initial = MmapSegment.create(initialPath, 0L, segmentSizeBytes);
+                    initial = MmapSegment.create(filesFacade, initialPath, 0L, segmentSizeBytes);
                 }
                 try {
-                    ringInProgress = new SegmentRing(initial, segmentSizeBytes);
+                    if (!memoryMode) {
+                        // Ordering is load-bearing for crash recovery:
+                        //   1. initial segment durable (header + dirent),
+                        //   2. manifest created (its own create fsyncs),
+                        //   3. flag stamped on the segment.
+                        // A crash after (1) recovers via the legacy
+                        // (manifest-less) path; after (2) via the manifest
+                        // path with an unflagged-but-valid empty active;
+                        // after (3) it is the steady state. Every window
+                        // self-heals without operator action.
+                        initial.syncHeader();
+                        if (filesFacade.fsyncDir(sfDir) != 0) {
+                            throw new MmapSegmentException(
+                                    "could not sync fresh SF segment directory " + sfDir);
+                        }
+                        initialManifest = SfManifest.create(filesFacade, sfDir, 0L, 0L);
+                        initial.markManifestRequired();
+                    }
+                    ringInProgress = new SegmentRing(initial, segmentSizeBytes, initialManifest);
+                    initialManifest = null;
                 } catch (Throwable t) {
                     initial.close();
+                    if (initialManifest != null) {
+                        initialManifest.close();
+                        SfManifest.removeFile(filesFacade, sfDir);
+                    }
                     if (initialPath != null) {
-                        Files.remove(initialPath);
+                        filesFacade.remove(initialPath);
                     }
                     throw t;
                 }
@@ -1247,17 +1280,24 @@ private static long segmentCleanupRank(String name) {
     }
 
     /**
-     * Unlinks every {@code .sfa} file under {@code dir}. Called only on
-     * clean shutdown when the ring confirms every published FSN has been
-     * acked — at that moment the slot has no recoverable work and the
-     * files are pure noise that would mislead the next sender's recovery.
+     * Unlinks every {@code .sfa} file under {@code dir}, then the SF
+     * manifest. Called only on clean shutdown when the ring confirms every
+     * published FSN has been acked — at that moment the slot has no
+     * recoverable work and the files are pure noise that would mislead the
+     * next sender's recovery.
      * <p>
-     * Removal runs in ascending segment order and STOPS at the first
-     * failure, so whatever survives (a failure here, or a crash mid-loop)
-     * is always a contiguous top slice of the ring: recovery's
-     * FSN-contiguity check still passes, and the retained ack watermark
-     * (== the final acked FSN == the highest frame on disk) still covers
-     * every surviving frame, so a successor replays nothing.
+     * Crash safety comes from a single durable manifest update committed
+     * BEFORE the first unlink: collapsing the boundaries to
+     * {@code head == active} declares every frame below the active base
+     * trimmed, so any subset of surviving files recovers cleanly (stale
+     * files are discarded, a surviving active recovers as the whole chain,
+     * zero files with a leftover manifest is the recognized drain window
+     * and recovers as EMPTY). Removal still runs in ascending segment order
+     * and STOPS at the first failure; the retained ack watermark (== the
+     * final acked FSN) covers every surviving frame, so a successor replays
+     * nothing. Legacy slots without a manifest keep the pre-manifest
+     * argument: ascending-order removal preserves a contiguous top slice
+     * that passes FSN-contiguity.
      *
      * @return {@code true} only when enumeration succeeded and every
      * {@code .sfa} file was confirmed removed — the caller keeps the ack
@@ -1300,13 +1340,54 @@ private static boolean unlinkAllSegmentFiles(FilesFacade filesFacade, String dir
             int byRank = Long.compare(segmentCleanupRank(a), segmentCleanupRank(b));
             return byRank != 0 ? byRank : a.compareTo(b);
         });
-        for (int i = 0, n = names.size(); i < n; i++) {
-            String path = dir + "/" + names.get(i);
-            if (!filesFacade.remove(path)) {
-                LOG.warn("Failed to unlink fully-acked segment {} on close; stopping so the "
-                        + "residual files stay a contiguous, watermark-covered range", path);
+        // Manifest-aware drain protocol. Before removing anything, durably
+        // collapse the committed boundaries to head == active. From that
+        // moment EVERY crash state is a valid recovery input:
+        //   - surviving data below the active base is "stale below head"
+        //     (acked by definition here) and is discarded by recovery;
+        //   - a surviving active-base segment recovers as the whole chain;
+        //   - empty spares are validated extras;
+        //   - zero segment files with the manifest still present is the
+        //     recognized drain window and recovers as EMPTY.
+        // Without this barrier, deleting the file the manifest calls "head"
+        // would make the next startup fail on a slot that lost nothing.
+        // The manifest itself is removed last, after every segment unlink.
+        SfManifest manifest = null;
+        try {
+            try {
+                manifest = SfManifest.open(filesFacade, dir);
+            } catch (Throwable t) {
+                LOG.warn("close-time unlink could not read the SF manifest in {}; leaving "
+                        + "all files for the next recovery", dir, t);
                 return false;
             }
+            if (manifest != null && !names.isEmpty()) {
+                try {
+                    long activeBase = manifest.activeBase();
+                    manifest.update(activeBase, activeBase);
+                } catch (Throwable t) {
+                    LOG.warn("close-time unlink could not commit drain boundaries in {}; "
+                            + "leaving all files for the next recovery", dir, t);
+                    return false;
+                }
+            }
+            for (int i = 0, n = names.size(); i < n; i++) {
+                String path = dir + "/" + names.get(i);
+                if (!filesFacade.remove(path)) {
+                    LOG.warn("Failed to unlink fully-acked segment {} on close; stopping so the "
+                            + "residual files stay a watermark-covered, manifest-consistent range", path);
+                    return false;
+                }
+            }
+        } finally {
+            if (manifest != null) {
+                manifest.close();
+            }
+        }
+        if (!SfManifest.removeFile(filesFacade, dir)) {
+            // Safe to proceed: a manifest with zero segment files is the
+            // recognized drain window and recovers as EMPTY.
+            LOG.warn("could not remove SF manifest in {} after full segment cleanup", dir);
         }
         return true;
     }