feat(ilp): cursor SF -- on-disk recovery + maxTotalBytes cap

Two pre-wiring additions on the cursor side that the upcoming
WebSocketSendQueue replacement will need.

1. SegmentRing.openExisting(sfDir, maxBytesPerSegment)
   Walks *.sfa files in the directory, opens each via
   MmapSegment.openExisting (which already validates header + scans
   torn tails), arranges by baseSeq, and returns a ring with the
   newest as active and the rest as sealed. Validates that the
   recovered segments form a contiguous FSN range -- a gap signals
   manual deletion or partial-write damage and aborts recovery
   rather than silently producing duplicate / missing FSNs after
   restart. Stray .sfa files with bad headers are skipped (logged-
   then-ignored), not fatal.

2. SegmentManager maxTotalBytes cap
   Manager tracks total bytes it has provisioned across all rings
   it serves. When provisioning a hot spare would exceed the cap,
   the manager skips the install and the requesting ring stays in
   BACKPRESSURE_NO_SPARE until ACK-driven trim frees space. Default
   is UNLIMITED_TOTAL_BYTES (no behavioural change for existing
   callers). Disk-full state is logged at WARN, throttled to once
   per 30s so a sustained-full state doesn't drown the log. Cap is
   approximate -- it counts only manager-provisioned segments, not
   the engine's initial active per ring (so the effective on-disk
   cap is maxTotalBytes + (rings * segmentSizeBytes)). Acceptable
   for a runaway-growth guard; documented in the constructor.

Also makes SegmentRing.sealedSegments mutation thread-safe via a
synchronized snapshot path that the I/O loop will use, and marks
SegmentRing.active volatile so cross-thread rotation publication
is correct without a lock.

10 new tests across SegmentRingTest + SegmentManagerTest covering:
recovery happy path, FSN-gap detection, bad-magic skip, cap blocks
provisioning, cap is released by ACK-driven trim. All 2020 tests in
the suite pass.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: bluestreak01

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

@@ -60,26 +60,68 @@
 public final class SegmentManager implements QuietCloseable {
 
     public static final long DEFAULT_POLL_NANOS = 1_000_000L; // 1 ms
+    public static final long DISK_FULL_LOG_THROTTLE_NANOS = 30_000_000_000L; // 30 s
+    public static final long UNLIMITED_TOTAL_BYTES = Long.MAX_VALUE;
     private static final Logger LOG = LoggerFactory.getLogger(SegmentManager.class);
 
     private final AtomicLong fileGeneration = new AtomicLong();
     private final Object lock = new Object();
+    private final long maxTotalBytes;
     private final long pollNanos;
     private final ObjList<RingEntry> rings = new ObjList<>();
     private final long segmentSizeBytes;
+    // Total bytes currently allocated across every segment owned by every
+    // registered ring (active + sealed + hot-spare). Manager-thread only —
+    // incremented when a spare is created, decremented when trim removes a
+    // segment. No lock needed because both operations happen on the manager
+    // thread inside serviceRing().
+    private long totalBytes;
+    private long lastDiskFullLogNs;
     private volatile boolean running;
     private Thread workerThread;
 
     public SegmentManager(long segmentSizeBytes) {
-        this(segmentSizeBytes, DEFAULT_POLL_NANOS);
+        this(segmentSizeBytes, DEFAULT_POLL_NANOS, UNLIMITED_TOTAL_BYTES);
     }
 
     public SegmentManager(long segmentSizeBytes, long pollNanos) {
+        this(segmentSizeBytes, pollNanos, UNLIMITED_TOTAL_BYTES);
+    }
+
+    /**
+     * Full constructor.
+     *
+     * @param segmentSizeBytes per-segment file size in bytes
+     * @param pollNanos how often the worker polls each registered ring;
+     *                  default {@link #DEFAULT_POLL_NANOS}
+     * @param maxTotalBytes upper bound on total bytes the manager will
+     *                      provision. When provisioning a hot spare would
+     *                      exceed this, the manager skips the install — the
+     *                      requesting ring stays in the
+     *                      {@link SegmentRing#BACKPRESSURE_NO_SPARE} state
+     *                      until ACK-driven trim frees space. Pass
+     *                      {@link #UNLIMITED_TOTAL_BYTES} to disable.
+     *                      <b>Approximation:</b> the cap counts only segments
+     *                      the manager itself provisioned. Each ring's
+     *                      initial active segment (created by the engine
+     *                      before the ring was registered) is "free" for
+     *                      cap purposes — so the effective on-disk cap is
+     *                      {@code maxTotalBytes + (rings × segmentSizeBytes)}.
+     *                      A 1-segment slop is acceptable for the cap's role
+     *                      (preventing runaway growth).
+     */
+    public SegmentManager(long segmentSizeBytes, long pollNanos, long maxTotalBytes) {
         if (segmentSizeBytes < MmapSegment.HEADER_SIZE + MmapSegment.FRAME_HEADER_SIZE + 1) {
             throw new IllegalArgumentException("segmentSizeBytes too small: " + segmentSizeBytes);
         }
+        if (maxTotalBytes < segmentSizeBytes) {
+            throw new IllegalArgumentException(
+                    "maxTotalBytes (" + maxTotalBytes + ") must allow at least one segment of "
+                            + segmentSizeBytes + " bytes");
+        }
         this.segmentSizeBytes = segmentSizeBytes;
         this.pollNanos = pollNanos;
+        this.maxTotalBytes = maxTotalBytes;
     }
 
     @Override
@@ -136,24 +178,41 @@ public synchronized void start() {
     }
 
     private void serviceRing(RingEntry e) {
-        // 1. Provision a hot spare if the ring needs one.
+        // 1. Provision a hot spare if the ring needs one AND we have headroom
+        //    under the disk-total cap. Cap check is per-tick; if we're capped
+        //    here, the ring stays in BACKPRESSURE_NO_SPARE until trim (step 2)
+        //    on this or a subsequent tick frees space. Logged at most once per
+        //    DISK_FULL_LOG_THROTTLE_NANOS so a sustained-disk-full state
+        //    doesn't drown the log.
         if (e.ring.needsHotSpare()) {
-            String path = nextSparePath(e.dir);
-            try {
-                // baseSeq is provisional — SegmentRing.appendOrFsn calls
-                // rebaseSeq() at rotation time to pin the real value. We
-                // pass the manager's best guess (nextSeqHint at this
-                // instant), which is fine since it's overwritten anyway.
-                MmapSegment spare = MmapSegment.create(path, e.ring.nextSeqHint(), segmentSizeBytes);
+            if (totalBytes + segmentSizeBytes > maxTotalBytes) {
+                long now = System.nanoTime();
+                if (now - lastDiskFullLogNs >= DISK_FULL_LOG_THROTTLE_NANOS) {
+                    LOG.warn("SF disk-full: cannot provision spare in {} "
+                                    + "(totalBytes={}, cap={}, segmentSize={}). "
+                                    + "Producer is backpressured until ACK-driven trim frees space.",
+                            e.dir, totalBytes, maxTotalBytes, segmentSizeBytes);
+                    lastDiskFullLogNs = now;
+                }
+            } else {
+                String path = nextSparePath(e.dir);
                 try {
-                    e.ring.installHotSpare(spare);
+                    // baseSeq is provisional — SegmentRing.appendOrFsn calls
+                    // rebaseSeq() at rotation time to pin the real value. We
+                    // pass the manager's best guess (nextSeqHint at this
+                    // instant), which is fine since it's overwritten anyway.
+                    MmapSegment spare = MmapSegment.create(path, e.ring.nextSeqHint(), segmentSizeBytes);
+                    try {
+                        e.ring.installHotSpare(spare);
+                        totalBytes += segmentSizeBytes;
+                    } catch (Throwable t) {
+                        spare.close();
+                        Files.remove(path);
+                        throw t;
+                    }
                 } catch (Throwable t) {
-                    spare.close();
-                    Files.remove(path);
-                    throw t;
+                    LOG.warn("Failed to provision hot spare in {} (will retry next tick)", e.dir, t);
                 }
-            } catch (Throwable t) {
-                LOG.warn("Failed to provision hot spare in {} (will retry next tick)", e.dir, t);
             }
         }
 
@@ -163,11 +222,13 @@ private void serviceRing(RingEntry e) {
             for (int i = 0, n = trim.size(); i < n; i++) {
                 MmapSegment s = trim.get(i);
                 String path = s.path();
+                long sz = s.sizeBytes();
                 try {
                     s.close();
                     if (!Files.remove(path)) {
                         LOG.warn("Failed to unlink trimmed segment {}", path);
                     }
+                    totalBytes -= sz;
                 } catch (Throwable t) {
                     LOG.warn("Failed to trim segment {}", path, t);
                 }

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

@@ -24,6 +24,7 @@
 
 package io.questdb.client.cutlass.qwp.client.sf.cursor;
 
+import io.questdb.client.std.Files;
 import io.questdb.client.std.ObjList;
 import io.questdb.client.std.QuietCloseable;
 
@@ -68,7 +69,10 @@ public final class SegmentRing implements QuietCloseable {
     // looks at sealedSegments after observing a higher ackedFsn, by which
     // point the producer thread's add to sealedSegments has retired.
     private final ObjList<MmapSegment> sealedSegments = new ObjList<>();
-    private MmapSegment active;
+    // active: written by producer (constructor + appendOrFsn rotation),
+    // read by I/O thread via getActive(). Volatile so the I/O thread sees
+    // rotations promptly and never observes a torn reference.
+    private volatile MmapSegment active;
     private volatile long ackedFsn = -1L;
     // hotSpare: written by segment manager (installHotSpare), read+cleared by
     // producer thread on rotation. Volatile so the producer sees fresh installs.
@@ -91,6 +95,110 @@ public SegmentRing(MmapSegment initialActive, long maxBytesPerSegment) {
         this.publishedFsn = nextSeq - 1;
     }
 
+    /**
+     * Recovers a ring from segments already on disk in {@code sfDir}. Used at
+     * sender startup when the user's previous session left durable but
+     * not-yet-acked frames behind. Walks every {@code *.sfa} file in the
+     * 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>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.
+     * <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.
+     */
+    public static SegmentRing openExisting(String sfDir, long maxBytesPerSegment) {
+        if (!Files.exists(sfDir)) {
+            return null;
+        }
+        ObjList<MmapSegment> opened = new ObjList<>();
+        long find = Files.findFirst(sfDir);
+        if (find == 0) {
+            return null;
+        }
+        try {
+            int rc = 1;
+            while (rc > 0) {
+                String name = Files.utf8ToString(Files.findName(find));
+                if (name != null && name.endsWith(".sfa") && !".".equals(name) && !"..".equals(name)) {
+                    String path = sfDir + "/" + name;
+                    try {
+                        opened.add(MmapSegment.openExisting(path));
+                    } catch (MmapSegmentException ignored) {
+                        // Stray file with the .sfa extension but bad header /
+                        // unreadable: skip rather than fail the recovery.
+                        // Logging is the engine's responsibility — SegmentRing
+                        // doesn't have a logger of its own.
+                    }
+                }
+                rc = Files.findNext(find);
+            }
+        } finally {
+            Files.findClose(find);
+        }
+        if (opened.size() == 0) {
+            return null;
+        }
+        // Sort by baseSeq ascending. ObjList lacks sort; do a simple selection
+        // sort — typical recovery is < 100 segments, O(n^2) is fine.
+        for (int i = 0, n = opened.size(); i < n; i++) {
+            int minIdx = i;
+            long minBase = opened.get(i).baseSeq();
+            for (int j = i + 1; j < n; j++) {
+                long b = opened.get(j).baseSeq();
+                if (b < minBase) {
+                    minBase = b;
+                    minIdx = j;
+                }
+            }
+            if (minIdx != i) {
+                MmapSegment tmp = opened.get(i);
+                opened.setQuick(i, opened.get(minIdx));
+                opened.setQuick(minIdx, tmp);
+            }
+        }
+        // Sanity: the recovered segments must form a contiguous FSN range.
+        // Detect gaps so a partial-write/manual-deletion mishap doesn't
+        // silently produce duplicate or missing FSNs after recovery.
+        for (int i = 1, n = opened.size(); i < n; i++) {
+            MmapSegment prev = opened.get(i - 1);
+            MmapSegment curr = opened.get(i);
+            long expected = prev.baseSeq() + prev.frameCount();
+            if (curr.baseSeq() != expected) {
+                // Close everything we've opened so the file handles don't leak.
+                for (int j = 0; j < n; j++) opened.get(j).close();
+                throw new MmapSegmentException(
+                        "FSN gap in recovered segments: prev baseSeq=" + prev.baseSeq()
+                                + " frameCount=" + prev.frameCount()
+                                + " expected next baseSeq=" + expected
+                                + " but got " + curr.baseSeq());
+            }
+        }
+        // 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.
+        int last = opened.size() - 1;
+        MmapSegment active = opened.get(last);
+        opened.remove(last);
+        SegmentRing ring = new SegmentRing(active, maxBytesPerSegment);
+        // Older segments become sealed in baseSeq order.
+        for (int i = 0, n = opened.size(); i < n; i++) {
+            ring.sealedSegments.add(opened.get(i));
+        }
+        return ring;
+    }
+
     /**
      * Highest FSN that the server has ACK'd. Read by the segment manager to
      * decide which sealed segments are safe to munmap + unlink.
@@ -137,7 +245,12 @@ public long appendOrFsn(long payloadAddr, int payloadLen) {
             // earlier guess at baseSeq is irrelevant.
             long actualBase = active.baseSeq() + active.frameCount();
             spare.rebaseSeq(actualBase);
-            sealedSegments.add(active);
+            // Mutate sealedSegments under the same monitor used by
+            // snapshotSealedSegments — the I/O thread reads through that
+            // path and must not see a half-resized ObjList.
+            synchronized (this) {
+                sealedSegments.add(active);
+            }
             active = spare;
             hotSpare = null;
             offset = active.tryAppend(payloadAddr, payloadLen);
@@ -179,11 +292,13 @@ public void close() {
      * when nothing is eligible (avoids ObjList allocation in the steady
      * state where most polls are no-ops).
      */
-    public ObjList<MmapSegment> drainTrimmable() {
+    public synchronized ObjList<MmapSegment> drainTrimmable() {
         long acked = ackedFsn;
         ObjList<MmapSegment> out = null;
         // Sealed segments are in baseSeq order, oldest first; once we hit one
         // that isn't fully acked, none of the later ones can be either.
+        // Synchronized so the I/O thread's snapshotSealedSegments() can't
+        // race against the remove(0) shuffling slots underneath it.
         while (sealedSegments.size() > 0) {
             MmapSegment s = sealedSegments.get(0);
             long lastSeq = s.baseSeq() + s.frameCount() - 1;
@@ -204,11 +319,44 @@ public MmapSegment getActive() {
         return active;
     }
 
-    /** Snapshot view of sealed segments (oldest first); for I/O thread to drain. */
+    /**
+     * Direct view of sealed segments (oldest first). NOT thread-safe — use
+     * only from the producer thread, or alongside a lock that excludes
+     * concurrent rotation. Cross-thread readers (typically the I/O loop)
+     * should use {@link #snapshotSealedSegments(MmapSegment[])} instead.
+     */
     public ObjList<MmapSegment> getSealedSegments() {
         return sealedSegments;
     }
 
+    /**
+     * Thread-safe snapshot of the current sealed-segment list. Copies
+     * references into the caller-supplied {@code target} array (oldest
+     * first, packed left). Returns the number of references copied. If
+     * {@code target} is too small, copies the first {@code target.length}
+     * references and returns {@code -1} as a signal that the caller needs
+     * to grow the buffer and retry.
+     * <p>
+     * Synchronized against rotation (producer's
+     * {@link #appendOrFsn} mutates {@code sealedSegments}). Cost is one
+     * monitor acquire/release per call, paid by the I/O loop at most once
+     * per tick — far below the cost of the actual {@code sendBinary} that
+     * the I/O loop is about to do.
+     */
+    public synchronized int snapshotSealedSegments(MmapSegment[] target) {
+        int n = sealedSegments.size();
+        if (n > target.length) {
+            for (int i = 0; i < target.length; i++) {
+                target[i] = sealedSegments.get(i);
+            }
+            return -1;
+        }
+        for (int i = 0; i < n; i++) {
+            target[i] = sealedSegments.get(i);
+        }
+        return n;
+    }
+
     /**
      * Segment manager pre-creates the next segment and parks it here. The
      * producer consumes the spare on its next rotation. Throws if a spare