Fail closed on unreadable SF segments and enumeration errors in recovery

SegmentRing.openExisting() previously skipped every per-file
MmapSegmentException and treated directory-enumeration failure
(findFirst < 0) as an empty store. The interior FSN-contiguity check
cannot expose a skipped segment at the lowest, highest, or sole
position -- the survivors stay mutually contiguous -- so recovery
silently retired the skipped segment's unacked frames (lowest/sole:
the engine seeds ackedFsn past them), minted duplicate FSNs
(highest: nextSeq resumes inside the skipped range), or -- for a
sole unreadable segment -- returned null, routing CursorSendEngine
onto the fresh-start path whose truncating openCleanRW
(O_CREAT|O_TRUNC / CREATE_ALWAYS) destroyed the only surviving
sf-initial.sfa.

Recovery now fails closed:
- an unreadable .sfa aborts recovery with MmapSegmentException
  naming the file and remediation options (restore, or move the
  file out to explicitly accept the loss);
- enumeration failure throws instead of returning null;
- quarantine-rename and empty-leftover-unlink failures throw
  instead of warn-and-continue;
- CursorSendEngine's fresh-start path refuses to create
  sf-initial.sfa over an existing file (defense in depth), checked
  before the stale-watermark unlink to preserve forensic state.

openExisting() returning null now strictly means "no recoverable
data existed". BackgroundDrainer already quarantines a slot
(markFailed + FAILED) when engine construction throws, so corrupt
orphan slots are preserved for postmortem rather than drained over.

New SegmentRingEdgeRecoveryTest pins the contract with sole /
lowest / highest unreadable-segment tests, an engine-level
no-truncation test, and a POSIX enumeration-failure test; all five
reproduced the data loss before the fix.

Author: bluestreak01

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

@@ -316,21 +316,36 @@ private CursorSendEngine(String sfDir, long segmentSizeBytes, SegmentManager man
                             recoveredCommitBoundaryFsn, publishedFsn);
                 }
             } else {
-                // Fresh start with no recovered segments. Any stale
-                // watermark from a prior fully-drained session refers
-                // to a lifecycle now gone -- unlink it before opening
-                // so the new session's first read() correctly reports
-                // INVALID (magic=0 on a freshly zero-filled file).
-                if (!memoryMode) {
-                    AckWatermark.removeOrphan(sfDir);
-                    watermarkInProgress = AckWatermark.open(sfDir);
-                }
+                // Fresh start with no recovered segments.
                 MmapSegment initial;
                 String initialPath = null;
                 if (memoryMode) {
                     initial = MmapSegment.createInMemory(0L, segmentSizeBytes);
                 } else {
                     initialPath = sfDir + "/sf-initial.sfa";
+                    // Defense in depth: recovery returning null now guarantees
+                    // no recognizable segment data remained (unreadable files
+                    // and enumeration failures fail closed; empty leftovers
+                    // were unlinked; corrupt-tailed ones quarantined). If a
+                    // file still exists at this path, some invariant broke --
+                    // and MmapSegment.create's openCleanRW uses truncating
+                    // semantics (O_CREAT|O_TRUNC / CREATE_ALWAYS), so creating
+                    // over it would silently destroy whatever it holds.
+                    // Checked BEFORE the watermark unlink below so a guard
+                    // trip preserves maximal forensic state for postmortem.
+                    if (Files.exists(initialPath)) {
+                        throw new MmapSegmentException(
+                                "refusing to overwrite existing " + initialPath
+                                        + " on fresh start -- recovery reported no data, yet the"
+                                        + " file exists; creating over it would truncate its"
+                                        + " contents");
+                    }
+                    // Any stale watermark from a prior fully-drained session
+                    // refers to a lifecycle now gone -- unlink it before
+                    // opening so the new session's first read() correctly
+                    // reports INVALID (magic=0 on a freshly zero-filled file).
+                    AckWatermark.removeOrphan(sfDir);
+                    watermarkInProgress = AckWatermark.open(sfDir);
                     initial = MmapSegment.create(initialPath, 0L, segmentSizeBytes);
                 }
                 try {

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

@@ -146,16 +146,24 @@ public SegmentRing(MmapSegment initialActive, long maxBytesPerSegment) {
      *       spare, then rotates onto it.</li>
      *   <li>All others become sealed segments awaiting ACK and trim.</li>
      * </ul>
-     * Returns {@code null} if the directory is empty or contains no
-     * recognizable {@code .sfa} files -- the caller should then construct a
-     * fresh ring with {@link #SegmentRing(MmapSegment, long)} and a freshly
-     * created initial segment.
+     * Returns {@code null} only when there is genuinely no data to recover:
+     * the directory does not exist, is empty, or contains only empty
+     * hot-spare leftovers (which are unlinked). The caller may then safely
+     * construct a fresh ring with {@link #SegmentRing(MmapSegment, long)} and
+     * a freshly created initial segment.
      * <p>
-     * Recovery is best-effort: a single bad-magic file is silently skipped
-     * (logged-then-ignored is the right call here; a stray unrelated file in
-     * the SF dir shouldn't take the whole sender down). A failure to open
-     * an otherwise-valid segment IS fatal -- the caller's data integrity
-     * depends on every segment being readable.
+     * Recovery fails closed: any {@code .sfa} file that cannot be opened --
+     * bad magic, bad header, unsupported version, short file, mmap rejection
+     * -- aborts recovery with {@link MmapSegmentException}, as does a failure
+     * to enumerate the directory. Skipping an unreadable file is never safe
+     * here: the interior FSN-contiguity check below cannot expose a skipped
+     * segment at the lowest, highest, or sole position (the survivors stay
+     * mutually contiguous), so recovery would silently retire the skipped
+     * segment's unacked frames, mint duplicate FSNs past a skipped tail, or
+     * -- for a sole skipped file -- route the caller onto the truncating
+     * fresh-start path that destroys the only surviving segment. An operator
+     * who wants to accept the loss moves the file out of the SF directory
+     * explicitly.
      */
     public static SegmentRing openExisting(String sfDir, long maxBytesPerSegment) {
         if (!Files.exists(sfDir)) {
@@ -164,9 +172,16 @@ public static SegmentRing openExisting(String sfDir, long maxBytesPerSegment) {
         ObjList<MmapSegment> opened = new ObjList<>();
         long find = Files.findFirst(sfDir);
         if (find < 0) {
-            LOG.warn("openExisting could not enumerate {} - treating as empty, "
-                    + "but this may indicate a permission or transient error", sfDir);
-            return null;
+            // Fail closed: an enumeration failure (permissions, transient
+            // I/O error) is NOT an empty store. Returning null here would
+            // route the caller onto the fresh-start path, whose truncating
+            // openCleanRW of sf-initial.sfa destroys any surviving data the
+            // JVM merely could not list.
+            throw new MmapSegmentException(
+                    "could not enumerate SF directory " + sfDir
+                            + " -- refusing to treat an unreadable directory as an empty"
+                            + " store (a fresh start would truncate sf-initial.sfa over"
+                            + " any surviving data); check directory permissions");
         }
         if (find == 0) {
             return null;
@@ -185,7 +200,30 @@ public static SegmentRing openExisting(String sfDir, long maxBytesPerSegment) {
                         String path = sfDir + "/" + name;
                         MmapSegment seg = null;
                         try {
-                            seg = MmapSegment.openExisting(path);
+                            try {
+                                seg = MmapSegment.openExisting(path);
+                            } catch (MmapSegmentException t) {
+                                // Fail closed: an unreadable .sfa in the slot is a
+                                // recovery failure, not a skippable curiosity. The
+                                // interior contiguity check below cannot expose a
+                                // skipped segment at the lowest, highest, or sole
+                                // position -- the survivors stay mutually
+                                // contiguous -- so skipping would silently retire
+                                // the segment's unacked frames (lowest/sole: the
+                                // engine seeds ackedFsn past them), mint duplicate
+                                // FSNs (highest: nextSeq resumes inside the skipped
+                                // range), or -- sole -- route the caller onto the
+                                // truncating fresh-start path that destroys the
+                                // only surviving file. The outer catch closes every
+                                // already-recovered segment before this propagates.
+                                throw new MmapSegmentException(
+                                        "recovery failed: unreadable segment file " + path
+                                                + " -- refusing to start with partial data (a skipped"
+                                                + " segment silently loses or duplicates frames). Restore"
+                                                + " the file from a copy, or move it out of the SF"
+                                                + " directory to explicitly accept the data loss",
+                                        t);
+                            }
                             // Filter out empty leftovers -- typically hot-spare
                             // segments the manager pre-allocated for a prior
                             // session that never got rotated into active. They
@@ -213,30 +251,39 @@ public static SegmentRing openExisting(String sfDir, long maxBytesPerSegment) {
                                 if (torn > 0) {
                                     int renameResult = Files.rename(path, path + ".corrupt");
                                     if (renameResult != 0) {
-                                        LOG.warn("openExisting: could not quarantine corrupt segment {}; "
-                                                + "the original file remains in place [rc={}]",
-                                                path, renameResult);
+                                        // Fail closed: with the quarantine rename
+                                        // failed, the corrupt file is still at its
+                                        // original path -- if it is the sole
+                                        // segment, the caller's fresh start would
+                                        // truncate it (sf-initial.sfa) or restart
+                                        // FSNs alongside it. Surface the error
+                                        // instead of proceeding over live data.
+                                        throw new MmapSegmentException(
+                                                "could not quarantine corrupt segment " + path
+                                                        + " to " + path + ".corrupt [rc=" + renameResult
+                                                        + "] -- refusing to continue recovery while the"
+                                                        + " corrupt file occupies its original path");
                                     }
                                 } else {
-                                    Files.remove(path);
+                                    if (!Files.remove(path)) {
+                                        // Same fail-closed reasoning: an empty
+                                        // leftover that cannot be unlinked would
+                                        // be truncated by the caller's fresh
+                                        // start (harmless for an empty file) --
+                                        // but unlink failing in a directory we
+                                        // just created files in means something
+                                        // is wrong (permissions changed, dir
+                                        // vanished); don't build on that ground.
+                                        throw new MmapSegmentException(
+                                                "could not unlink empty leftover segment " + path
+                                                        + " -- refusing to continue recovery; check"
+                                                        + " directory permissions");
+                                    }
                                 }
                             } else {
                                 opened.add(seg);
                                 seg = null;
                             }
-                        } catch (MmapSegmentException t) {
-                            // Per-file data error (bad magic, bad header,
-                            // unsupported version, mmap rejection on this one
-                            // file). Don't take down recovery for one corrupt
-                            // .sfa -- log and skip so siblings still recover.
-                            // Resource exhaustion (OOM) and programmer errors
-                            // (IOOBE) deliberately propagate to the outer
-                            // catch, which closes every already-recovered
-                            // segment and rethrows: continuing the loop after
-                            // an OOM would just fail again on the next file
-                            // while silently leaking the segments we managed
-                            // to recover before it.
-                            LOG.warn("openExisting: skipping {} -- {}", path, t.toString());
                         } finally {
                             // Close any seg whose ownership wasn't transferred
                             // (either to opened, or via the empty-branch close

core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/MmapSegmentRecoveryFaultTest.java

@@ -208,12 +208,15 @@ public void testRecoverySurvivesPayloadReachingUnbackedPage() throws Exception {
     }
 
     /**
-     * An unbacked page-zero header must produce the per-file
-     * {@link MmapSegmentException} that SegmentRing skips, not poison recovery
-     * of valid sibling files.
+     * An unbacked page-zero header must produce a synchronous
+     * {@link MmapSegmentException}, not a SIGBUS/undefined mapping.
+     * SegmentRing.openExisting fails recovery closed on this exception --
+     * a segment with an unreadable header may hold recoverable frames, and
+     * silently skipping it at the lowest/highest/sole position loses or
+     * duplicates FSNs (see SegmentRingEdgeRecoveryTest).
      */
     @Test
-    public void testUnbackedHeaderPageIsSkippableNotFatal() throws Exception {
+    public void testUnbackedHeaderPageFailsSynchronously() throws Exception {
         TestUtils.assertMemoryLeak(() -> {
             final String path = tmpDir + "/seg-unbacked-header.sfa";
             writeSegment(path, 3L, new int[]{64});
@@ -223,8 +226,9 @@ public void testUnbackedHeaderPageIsSkippableNotFatal() throws Exception {
                 MmapSegment.openExisting(path).close();
                 fail("expected MmapSegmentException for an unbacked header page");
             } catch (MmapSegmentException expected) {
-                // ok -- SegmentRing's per-file catch skips just this file
-                // instead of aborting recovery of the whole slot.
+                // ok -- surfaces synchronously; SegmentRing.openExisting
+                // fails recovery closed on it rather than silently skipping
+                // a file that may hold recoverable frames.
             }
         });
     }
@@ -331,11 +335,12 @@ public void testSuccessfulRecoveryClosesThroughFacadeExactlyOnce() throws Except
     }
 
     /**
-     * A header short-read produces the per-file exception that SegmentRing
-     * skips; recovery never maps the stale reported length.
+     * A header short-read produces a synchronous per-file exception on which
+     * SegmentRing.openExisting fails recovery closed; recovery never maps
+     * the stale reported length.
      */
     @Test
-    public void testHeaderShortReadIsSkippable() throws Exception {
+    public void testHeaderShortReadFailsSynchronously() throws Exception {
         TestUtils.assertMemoryLeak(() -> {
             final String path = tmpDir + "/seg-mappasteof-header.sfa";
             final long page = Files.PAGE_SIZE;