fix(qwp): recover SF segments with positioned reads

Validate persisted segment bytes through bounded positioned reads before mmap. Fail closed on read and size instability, preserve corruption quarantine semantics, and cover sparse tails, buffer boundaries, and recovery I/O faults.

Author: bluestreak01

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

@@ -27,9 +27,9 @@
 /**
  * Positively-identified segment corruption: the file's own bytes prove it is
  * not (or no longer) a readable SF segment — truncated below the fixed header,
- * wrong magic, a negative {@code baseSeq}, or an unreadable (unbacked/torn)
- * header page. Distinct from its parent {@link MmapSegmentException}, which
- * recovery treats as an <b>operational</b> failure (open/mmap error on a file
+ * wrong magic, or a negative {@code baseSeq}. Distinct from its parent
+ * {@link MmapSegmentException}, which recovery treats as an <b>operational</b>
+ * failure (open/read/mmap error on a file
  * whose contents may be perfectly intact) and must therefore be fatal.
  * <p>
  * Recovery quarantines corruption (rename to {@code <name>.corrupt}) and

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

@@ -28,10 +28,10 @@
  * Hard failure of the MmapSegment layer — bad header, mmap rejection, file
  * too short for header, etc. Indicates the segment is unusable, not that
  * the disk is full (the latter surfaces as backpressure on the producer
- * via {@link io.questdb.client.cutlass.qwp.client.LineSenderException}).
+ * via {@link io.questdb.client.cutlass.line.LineSenderException}).
  * <p>
  * Recovery distinguishes two flavors: this base type marks <b>operational</b>
- * failures (open/mmap/enumeration errors — the file's contents may be fine,
+ * failures (open/read/mmap/enumeration errors — the file's contents may be fine,
  * so recovery must fail closed), while the {@link MmapSegmentCorruptionException}
  * subtype marks <b>positively-identified corruption</b> in the file's own
  * bytes, which recovery may quarantine instead of aborting.

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

@@ -226,16 +226,16 @@ static Recovery recover(
 
         ObjList<MmapSegment> all = new ObjList<>();
         // Files whose own bytes prove corruption (bad magic, sub-header size,
-        // negative baseSeq, unreadable header page). They are excluded from
-        // the chain and quarantined to <name>.corrupt — but only AFTER the
-        // surviving chain validates (or resolves to EMPTY), so a failed
-        // recovery never mutates the slot. Whether a quarantined file was
-        // load-bearing is decided by the manifest-boundary / contiguity
-        // checks below, not by the skip itself. Operational open errors
-        // (EMFILE, EACCES, mmap rejection, unsupported version) are NOT in
-        // this bucket: they throw the plain MmapSegmentException type and
-        // abort recovery, because the underlying file may be perfectly
-        // intact and silently dropping it could lose durable frames.
+        // negative baseSeq). They are excluded from the chain and quarantined
+        // to <name>.corrupt — but only AFTER the surviving chain validates (or
+        // resolves to EMPTY), so a failed recovery never mutates the slot.
+        // Whether a quarantined file was load-bearing is decided by the
+        // manifest-boundary / contiguity checks below, not by the skip itself.
+        // Operational open/stat/read/mmap errors, observed size instability,
+        // and unsupported versions are NOT in this bucket: they throw the
+        // plain MmapSegmentException type and abort recovery, because the
+        // underlying file may be perfectly intact and silently dropping it
+        // could lose durable frames.
         ObjList<String> corruptPaths = null;
         SfManifest manifest = null;
         try {

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

@@ -89,21 +89,18 @@ default int fsyncDir(String dir) {
         return Files.fsyncDir(dir);
     }
 
+    /**
+     * Returns the current byte length of the file referenced by open descriptor
+     * {@code fd}, or a negative value when the descriptor cannot be statted.
+     */
     long length(int fd);
 
     /**
-     * Stat length of the file at {@code path}, in bytes. Default delegates to
-     * {@link Files#length(String)}.
-     *
-     * <p>Test injection point: {@code MmapSegment.openExisting} maps the file to
-     * this length and scans straight out of the mapping, so a wrapping facade
-     * that returns a value <em>larger</em> than the real file makes the mapping
-     * extend past end-of-file. A read of a page beyond real EOF raises SIGBUS on
-     * <em>every</em> filesystem (which HotSpot translates to a catchable
-     * {@code InternalError} at an {@code Unsafe} intrinsic site) — the same fault
-     * a genuinely unbacked/sparse page raises on ZFS, but reproduced
-     * deterministically on ext4/xfs too. That is what lets recovery's mmap-fault
-     * guard be regression-tested on any CI runner rather than only on ZFS.
+     * Stat length of the file at {@code path}, in bytes.
+     * {@link DefaultFilesFacade} delegates to {@link Files#length(String)}.
+     * Code that already owns an open descriptor should prefer
+     * {@link #length(int)} so path replacement cannot make the stat refer to a
+     * different file.
      */
     long length(String path);
 
@@ -152,6 +149,13 @@ default int openRWExclusive(long pathPtr) {
      */
     long length(long pathPtr);
 
+    /**
+     * Reads up to {@code len} bytes from the absolute file {@code offset} into
+     * native memory at {@code addr}, without changing the descriptor position.
+     * A positive result may be shorter than requested and callers that require
+     * the whole range must retry. For a non-zero request, {@code 0} means EOF
+     * or no progress and a negative result indicates an operating-system error.
+     */
     long read(int fd, long addr, long len, long offset);
 
     boolean remove(String path);