Fix SIGBUS risk when appending into recovered sparse SF segments

Recovery mmap'd the segment file's complete logical length RW and left
appendCursor at lastGoodOffset, exposing the unvalidated tail as
writable capacity. If that tail was sparse/unbacked (fallocate
reservations do not survive external sparsification, and
FilesFacade.allocate cannot retroactively fill holes below EOF), the
next producer append stored straight into the hole — which under
ENOSPC raises SIGBUS and aborts the JVM.

MmapSegment.openExisting now maps only the validated prefix
[0, lastGoodOffset), making recovered segments born full: the first
append backpressures until the segment manager installs a hot spare
(created with genuine block reservation and clean ENOSPC failure),
then rotates onto it. Consumers are unaffected — they only read below
publishedOffset(), which equals lastGoodOffset.

Because a recovered segment's mapped size no longer reflects its
on-disk footprint, sf_max_total_bytes accounting (register seed and
trim decrement) now uses the new MmapSegment.fileBytes() — the file's
logical length — instead of sizeBytes().

Regression tests: recovered sparse-tail segment must be born full and
refuse tryAppend; a ring recovered around a sparse-tail active must
backpressure the first append and resume with a contiguous FSN after
spare rotation.

Author: bluestreak01

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

@@ -70,6 +70,11 @@ public final class MmapSegment implements QuietCloseable {
     private final FilesFacade filesFacade;
     private final String path;
     private final long sizeBytes;
+    // fileBytes: logical size of the backing file — the segment's on-disk
+    // footprint for sf_max_total_bytes accounting. Equals sizeBytes for
+    // created and memory-backed segments; for recovered segments it can
+    // exceed sizeBytes because recovery maps only the validated prefix.
+    private final long fileBytes;
     // memoryBacked: true when the segment buffer lives in malloc'd native
     // memory rather than an mmap'd file. The "non-SF async" path uses
     // memory-backed segments — same cursor architecture, no disk involvement.
@@ -104,13 +109,14 @@ public final class MmapSegment implements QuietCloseable {
     private final long tornTailBytes;
 
     private MmapSegment(String path, int fd, long mmapAddress, long sizeBytes,
-                        long baseSeq, long initialCursor, long frameCount,
+                        long fileBytes, long baseSeq, long initialCursor, long frameCount,
                         boolean memoryBacked, long tornTailBytes, FilesFacade filesFacade) {
         this.path = path;
         this.filesFacade = filesFacade;
         this.fd = fd;
         this.mmapAddress = mmapAddress;
         this.sizeBytes = sizeBytes;
+        this.fileBytes = fileBytes;
         this.baseSeq = baseSeq;
         this.appendCursor = initialCursor;
         this.publishedCursor = initialCursor;
@@ -200,7 +206,7 @@ public static MmapSegment create(FilesFacade ff, long pathPtr, String displayPat
             Unsafe.getUnsafe().putLong(addr + 8, baseSeq);
             Unsafe.getUnsafe().putLong(addr + 16, Os.currentTimeMicros());
             return new MmapSegment(
-                    displayPath, fd, addr, sizeBytes, baseSeq, HEADER_SIZE, 0, false, 0L, ff);
+                    displayPath, fd, addr, sizeBytes, sizeBytes, baseSeq, HEADER_SIZE, 0, false, 0L, ff);
         } catch (Throwable t) {
             if (addr != Files.FAILED_MMAP_ADDRESS) {
                 Files.munmap(addr, sizeBytes, MemoryTag.MMAP_DEFAULT);
@@ -238,21 +244,30 @@ public static MmapSegment createInMemory(long baseSeq, long sizeBytes) {
             Unsafe.getUnsafe().putLong(addr + 8, baseSeq);
             Unsafe.getUnsafe().putLong(addr + 16, Os.currentTimeMicros());
             return new MmapSegment(
-                    null, -1, addr, sizeBytes, baseSeq, HEADER_SIZE, 0, true, 0L, null);
+                    null, -1, addr, sizeBytes, sizeBytes, baseSeq, HEADER_SIZE, 0, true, 0L, null);
         } catch (Throwable t) {
             Unsafe.free(addr, sizeBytes, MemoryTag.NATIVE_DEFAULT);
             throw t;
         }
     }
 
     /**
-     * Opens an existing segment file for recovery. mmaps it RW, validates the
-     * header magic / version, then scans frames forward verifying each CRC.
-     * The first bad CRC (or a frame whose declared length runs past the file
-     * end) is treated as a torn tail; both cursors are positioned at the
-     * start of that frame. Returns the segment ready for further appends.
+     * Opens an existing segment file for recovery. Validates the header magic
+     * / version, then scans frames forward verifying each CRC. The first bad
+     * CRC (or a frame whose declared length runs past the file end) is treated
+     * as a torn tail; both cursors are positioned at the start of that frame.
      * Throws {@link MmapSegmentException} on header validation failure.
      * <p>
+     * The returned segment maps ONLY the validated prefix
+     * {@code [0, lastGoodOffset)} and is therefore born full:
+     * {@link #tryAppend} returns -1 and the caller rotates onto a freshly
+     * created segment before the next append. The tail past the last good
+     * frame is never mapped — it may be sparse/unbacked ({@code create()}'s
+     * block reservation does not survive external sparsification, and
+     * {@link FilesFacade#allocate} cannot retroactively fill holes below EOF),
+     * and a store into an unbacked page under ENOSPC raises SIGBUS, aborting
+     * the JVM.
+     * <p>
      * If recovery observes a torn tail (the bytes at the bail-out position
      * are non-zero, indicating an attempted-but-failed frame write rather
      * than clean unwritten space), a {@code WARN} is emitted with the byte
@@ -303,11 +318,23 @@ public static MmapSegment openExisting(FilesFacade ff, String path) {
                         "short read before stable file EOF during recovery: " + path
                                 + " size=" + fileSize);
             }
-            mapSize = Math.min(fileSize, finalFileSize);
-            if (mapSize < HEADER_SIZE) {
+            long logicalSize = Math.min(fileSize, finalFileSize);
+            if (logicalSize < HEADER_SIZE) {
                 throw new MmapSegmentException("file shorter than header after recovery scan: "
-                        + path + " size=" + mapSize);
+                        + path + " size=" + logicalSize);
             }
+            // Map ONLY the validated prefix [0, lastGoodOffset). The tail past
+            // the last good frame may be sparse/unbacked, and a store into an
+            // unbacked page under ENOSPC raises SIGBUS and aborts the JVM.
+            // FilesFacade.allocate cannot make that tail safe at recovery time
+            // (it reserves [currentSize, target) only — pre-existing holes
+            // below EOF are not retroactively filled). Mapping through
+            // lastGoodOffset makes the recovered segment born full
+            // (capacityRemaining() == 0), so the producer rotates onto a
+            // freshly created — and genuinely block-reserved — segment before
+            // the next append. Consumers are unaffected: they only read below
+            // publishedOffset(), which equals lastGoodOffset.
+            mapSize = recovered.lastGoodOffset;
             addr = Files.mmap(fd, mapSize, 0, Files.MAP_RW, MemoryTag.MMAP_DEFAULT);
             if (addr == Files.FAILED_MMAP_ADDRESS) {
                 throw new MmapSegmentException("mmap failed for " + path);
@@ -317,23 +344,24 @@ public static MmapSegment openExisting(FilesFacade ff, String path) {
                     fd,
                     addr,
                     mapSize,
+                    logicalSize,
                     recovered.baseSeq,
                     recovered.lastGoodOffset,
                     recovered.frameCount,
                     false,
-                    Math.min(recovered.tornTailBytes, mapSize - recovered.lastGoodOffset),
+                    Math.min(recovered.tornTailBytes, logicalSize - recovered.lastGoodOffset),
                     ff
             );
             if (segment.tornTailBytes() > 0) {
                 LOG.warn("SF segment {}: torn tail of {} bytes at offset {} "
                                 + "(file size {}, frames recovered {}). "
-                                + "Recovery will overwrite this region on next append; "
+                                + "The torn region is excluded from the recovered mapping and "
                                 + "frames past the tear (if any) are discarded. "
                                 + "Investigate disk health or unexpected writer crash.",
                         path,
                         segment.tornTailBytes(),
                         segment.publishedOffset(),
-                        mapSize,
+                        logicalSize,
                         segment.frameCount());
             }
             // Ownership of fd and addr transfers to the returned segment.
@@ -443,6 +471,19 @@ public long sizeBytes() {
         return sizeBytes;
     }
 
+    /**
+     * Logical size of the backing file in bytes — the segment's on-disk
+     * footprint for {@code sf_max_total_bytes} cap accounting. Equals
+     * {@link #sizeBytes()} for created and memory-backed segments; for
+     * recovered segments it can exceed {@link #sizeBytes()} because recovery
+     * maps only the validated prefix (SIGBUS hardening in
+     * {@link #openExisting}) while the file keeps its full logical length
+     * until trim unlinks it.
+     */
+    public long fileBytes() {
+        return fileBytes;
+    }
+
     /**
      * Appends one frame: writes {@code [crc32c | u32 payloadLen | payload]}
      * starting at the current append cursor, then advances both cursors

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

@@ -744,7 +744,11 @@ private void serviceRing0(RingEntry e) {
             trim = e.ring.drainTrimmable();
             if (registered && trim != null) {
                 for (int i = 0, n = trim.size(); i < n; i++) {
-                    totalBytes -= trim.get(i).sizeBytes();
+                    // fileBytes(), not sizeBytes(): must mirror the register-time
+                    // seed (SegmentRing.totalSegmentBytes), which accounts the
+                    // on-disk footprint — recovered segments map only their
+                    // validated prefix but occupy the full file length on disk.
+                    totalBytes -= trim.get(i).fileBytes();
                 }
             }
         }

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

@@ -139,8 +139,11 @@ public SegmentRing(MmapSegment initialActive, long maxBytesPerSegment) {
      * directory, opens each via {@link MmapSegment#openExisting}, and
      * arranges them by baseSeq:
      * <ul>
-     *   <li>Highest-baseSeq segment becomes the active (further appends land
-     *       there until it fills, at which point normal rotation kicks in).</li>
+     *   <li>Highest-baseSeq segment becomes the active. Recovered segments
+     *       are mapped only through their validated prefix (SIGBUS hardening
+     *       in {@link MmapSegment#openExisting}) and are therefore born full:
+     *       the first append backpressures until the manager installs a hot
+     *       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
@@ -291,10 +294,12 @@ public static SegmentRing openExisting(String sfDir, long maxBytesPerSegment) {
                                     + " check disk health");
                 }
             }
-            // The newest segment becomes the active. Even if it's full, that's OK:
-            // the next appendOrFsn returns BACKPRESSURE_NO_SPARE, the manager
-            // installs a hot spare, the producer rotates. Same fast path as a
-            // mid-life ring.
+            // The newest segment becomes the active. Recovered segments are
+            // born full (mapped only through the validated prefix -- see the
+            // SIGBUS hardening in MmapSegment.openExisting), so the next
+            // appendOrFsn returns BACKPRESSURE_NO_SPARE, the manager installs
+            // a hot spare, the producer rotates. Same fast path as a mid-life
+            // ring whose active just filled.
             int last = opened.size() - 1;
             MmapSegment active = opened.get(last);
             opened.remove(last);
@@ -682,21 +687,25 @@ public synchronized int snapshotSealedSegments(MmapSegment[] target) {
     }
 
     /**
-     * Total mmap'd bytes the ring currently owns: active + hot spare (if
+     * Total on-disk bytes the ring currently owns: active + hot spare (if
      * installed) + every sealed segment. Used by {@code SegmentManager}
      * to seed its {@code totalBytes} accounting at register time and to
-     * reverse the contribution at deregister time. Synchronized against
-     * rotation so we never read a half-resized sealed list.
+     * reverse the contribution at deregister time. Uses
+     * {@link MmapSegment#fileBytes()} — the file's logical length — rather
+     * than the mapped size, because recovered segments map only their
+     * validated prefix while the file keeps its full length until trim
+     * unlinks it. Synchronized against rotation so we never read a
+     * half-resized sealed list.
      */
     public synchronized long totalSegmentBytes() {
         long total = 0L;
         MmapSegment a = active;
-        if (a != null) total += a.sizeBytes();
+        if (a != null) total += a.fileBytes();
         MmapSegment hs = hotSpare;
-        if (hs != null) total += hs.sizeBytes();
+        if (hs != null) total += hs.fileBytes();
         for (int i = 0, n = sealedSegments.size(); i < n; i++) {
             MmapSegment s = sealedSegments.get(i);
-            if (s != null) total += s.sizeBytes();
+            if (s != null) total += s.fileBytes();
         }
         return total;
     }

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

@@ -26,6 +26,7 @@
 
 import io.questdb.client.cutlass.qwp.client.sf.cursor.MmapSegment;
 import io.questdb.client.cutlass.qwp.client.sf.cursor.MmapSegmentException;
+import io.questdb.client.cutlass.qwp.client.sf.cursor.SegmentRing;
 import io.questdb.client.std.Files;
 import io.questdb.client.std.FilesFacade;
 import io.questdb.client.std.MemoryTag;
@@ -91,6 +92,84 @@ public void testRecoveryKeepsFramesBeforeUnbackedTail() throws Exception {
         });
     }
 
+    /**
+     * Regression for the recovered-sparse-tail SIGBUS: recovery used to mmap
+     * the file's complete logical size RW and leave the sparse tail as
+     * writable capacity, so the next append stored straight into the unbacked
+     * hole — which under ENOSPC raises SIGBUS and aborts the JVM.
+     * {@link FilesFacade#allocate} cannot retroactively fill holes below EOF,
+     * so no recovery-time reservation can make that tail safe; instead the
+     * fix maps only the validated prefix. The recovered segment must be born
+     * full and refuse appends, forcing rotation onto a freshly created,
+     * genuinely block-reserved segment.
+     */
+    @Test
+    public void testRecoveredSparseSegmentIsBornFullAndRefusesAppends() throws Exception {
+        TestUtils.assertMemoryLeak(() -> {
+            final String path = tmpDir + "/seg-sparse-no-append.sfa";
+            final long page = Files.PAGE_SIZE;
+            final int payloadLen = (int) (page - MmapSegment.HEADER_SIZE - MmapSegment.FRAME_HEADER_SIZE);
+            long boundary = writeSegment(path, 5L, new int[]{payloadLen});
+            assertEquals("frame must fill exactly one page", page, boundary);
+            punchSparseTail(path, page);
+
+            long buf = Unsafe.malloc(16, MemoryTag.NATIVE_DEFAULT);
+            try (MmapSegment seg = MmapSegment.openExisting(path)) {
+                assertEquals("recovered data must remain readable", 1L, seg.frameCount());
+                assertEquals("scan must stop at the unbacked-page boundary", page, seg.publishedOffset());
+                assertEquals("mapping must cover only the validated prefix", page, seg.sizeBytes());
+                assertTrue("recovered segment must be born full", seg.isFull());
+                assertEquals("no payload capacity may remain over the sparse tail",
+                        0L, seg.capacityRemaining());
+                assertEquals("append into the unbacked tail must be refused",
+                        -1L, seg.tryAppend(buf, 16));
+            } finally {
+                Unsafe.free(buf, 16, MemoryTag.NATIVE_DEFAULT);
+            }
+        });
+    }
+
+    /**
+     * Recovery-to-producer path for the same regression: a ring recovered
+     * around a sparse-tail active must backpressure the first append (never
+     * touching the hole) and resume cleanly — with a contiguous FSN sequence
+     * — once a genuinely allocated hot spare is installed and rotated in.
+     */
+    @Test
+    public void testRecoveredRingRotatesBeforeAppendingPastSparseTail() throws Exception {
+        TestUtils.assertMemoryLeak(() -> {
+            final String path = tmpDir + "/sf-0.sfa";
+            final long page = Files.PAGE_SIZE;
+            final int payloadLen = (int) (page - MmapSegment.HEADER_SIZE - MmapSegment.FRAME_HEADER_SIZE);
+            long boundary = writeSegment(path, 0L, new int[]{payloadLen});
+            assertEquals("frame must fill exactly one page", page, boundary);
+            punchSparseTail(path, page);
+
+            long buf = Unsafe.malloc(16, MemoryTag.NATIVE_DEFAULT);
+            try {
+                try (SegmentRing recovered = SegmentRing.openExisting(tmpDir, SEGMENT_BYTES)) {
+                    assertTrue("ring must recover from the on-disk segment", recovered != null);
+                    assertEquals(1L, recovered.getActive().frameCount());
+                    assertEquals(1L, recovered.nextSeqHint());
+                    // First append must NOT write into the sparse tail — the
+                    // born-full active backpressures until a spare arrives.
+                    assertEquals(SegmentRing.BACKPRESSURE_NO_SPARE,
+                            recovered.appendOrFsn(buf, 16));
+                    // create() pre-allocates real blocks, so the rotated-in
+                    // spare is safe to store into.
+                    recovered.installHotSpare(
+                            MmapSegment.create(tmpDir + "/sf-1.sfa", 1L, SEGMENT_BYTES));
+                    assertEquals("append must resume with a contiguous FSN",
+                            1L, recovered.appendOrFsn(buf, 16));
+                    assertEquals("append must land in the rotated-in spare",
+                            1L, recovered.getActive().baseSeq());
+                }
+            } finally {
+                Unsafe.free(buf, 16, MemoryTag.NATIVE_DEFAULT);
+            }
+        });
+    }
+
     /**
      * The harder case: a frame whose 8-byte header sits on a backed page but
      * whose payload reaches into the unbacked hole (a torn write leaves a real

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

@@ -294,7 +294,8 @@ public void testOpenExistingRecoversActivePlusSealed() throws Exception {
             long buf = Unsafe.malloc(16, MemoryTag.NATIVE_DEFAULT);
             try {
                 // Write three segments with FSN ranges 0..3, 4..7, 8..9 (last
-                // partially full so the recovered ring has appendable room).
+                // partially full; recovery still maps it born-full — see the
+                // SIGBUS hardening in MmapSegment.openExisting).
                 MmapSegment s0 = MmapSegment.create(tmpDir + "/r0.sfa", 0, segSize);
                 for (int i = 0; i < 4; i++) s0.tryAppend(buf, 16);
                 s0.close();
@@ -319,8 +320,17 @@ public void testOpenExistingRecoversActivePlusSealed() throws Exception {
                     assertEquals(4, recovered.getSealedSegments().get(1).baseSeq());
                     // nextSeq must continue past the recovered frames.
                     assertEquals(10, recovered.nextSeqHint());
-                    // Further appends land into the active and assign FSN 10.
+                    // Recovered segments are mapped only through their validated
+                    // prefix and are born full: the first append backpressures
+                    // (never writing into the potentially unbacked tail) until a
+                    // spare is installed, then rotates and assigns FSN 10.
+                    assertEquals(SegmentRing.BACKPRESSURE_NO_SPARE,
+                            recovered.appendOrFsn(buf, 16));
+                    recovered.installHotSpare(
+                            MmapSegment.create(tmpDir + "/r3.sfa", 10, segSize));
                     assertEquals(10, recovered.appendOrFsn(buf, 16));
+                    assertEquals("append must land in the rotated-in spare",
+                            10, recovered.getActive().baseSeq());
                 }
             } finally {
                 Unsafe.free(buf, 16, MemoryTag.NATIVE_DEFAULT);