perf(qwp): skip proven-durable sealed prefix in periodic SF sync

servicePeriodicSync copied and scanned the whole live sealed list under the ring monitor every tick and called syncPublished() on each. But a segment is made durable before it is sealed (the requestSyncBeforeRotation gate), so every sealed barrier early-returns -- the copy was O(live-sealed) monitor-held work that grows without bound under a producer-outpaces-drain backlog.

Track a monotonic durability frontier (firstNonDurableSealed) and copy only [frontier, size) + active via copyPendingSyncSegments. The frontier only advances past segments observed durable and durability never regresses, so it is a conservative lower bound that can never skip a segment still needing a barrier; recovery-resumed non-durable sealed segments are covered because it starts at 0. Maintained under the ring monitor and shifted with sealed-list compaction, so steady-state copy is O(1).

Adds a frontier invariant assert plus tests for the durable-prefix skip and the compaction index-shift. Also folds in minor local cleanups (Arrays.fill, RetainedSegmentMembership lambdas, assertTrue).

Author: bluestreak01

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

@@ -34,6 +34,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.util.Arrays;
 import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
@@ -1094,9 +1095,7 @@ private boolean serviceRing0(RingEntry e) {
                 batchSize = e.ring.stagePendingTrims(
                         trimBatch, MAX_TRIMS_PER_RING_PASS, durableAck);
             } catch (Throwable t) {
-                for (int i = 0; i < trimBatch.length; i++) {
-                    trimBatch[i] = null;
-                }
+                Arrays.fill(trimBatch, null);
                 recordTrimFailure(e, TRIM_RETRY_UNLINK, now, t);
                 return false;
             }
@@ -1219,17 +1218,20 @@ private void servicePeriodicSync(RingEntry e, long now) {
             return;
         }
         try {
-            e.ring.copyLiveSegmentsForSync(syncScratch);
+            e.ring.copyPendingSyncSegments(syncScratch);
             for (int i = 0, n = syncScratch.size(); i < n; i++) {
                 syncScratch.getQuick(i).syncPublished();
             }
             e.ring.clearSyncRequestIfActiveDurable();
-            // The pass above covered every live segment's published range, and
-            // a failed barrier re-dirties its range under an mlock pin (see
-            // MmapSegment.syncPublished), so a success here is a genuine
-            // re-persist -- not a vacuous retry over pages a failed writeback
-            // marked clean (fsyncgate). Unlatch so a transient disk fault
-            // doesn't permanently brick the producer.
+            // The pass above covered every not-yet-durable live range -- the
+            // proven-durable sealed prefix is skipped precisely because its
+            // syncPublished would early-return (see copyPendingSyncSegments),
+            // so any latched failure necessarily belongs to a range we just
+            // re-barriered. A failed barrier re-dirties its range under an
+            // mlock pin (see MmapSegment.syncPublished), so a success here is a
+            // genuine re-persist -- not a vacuous retry over pages a failed
+            // writeback marked clean (fsyncgate). Unlatch so a transient disk
+            // fault doesn't permanently brick the producer.
             e.ring.clearDurabilityFailure();
             e.nextDataSyncNanos = now + e.syncIntervalNanos;
             if (e.syncFailureLogged) {

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

@@ -100,6 +100,19 @@ public final class SegmentRing implements QuietCloseable {
     // Logical head of sealedSegments. Head removal nulls one entry and advances
     // this index; occasional compaction bounds unused prefix slots.
     private int sealedHead;
+    // Frontier index into sealedSegments: every sealed segment in
+    // [sealedHead, firstNonDurableSealed) has been proven durable by an earlier
+    // periodic pass and never needs re-scanning -- publishedCursor is frozen at
+    // seal and durableCursor only advances, so durability never regresses. The
+    // periodic sync (copyPendingSyncSegments) skips this proven-durable prefix,
+    // keeping the steady-state copy-under-monitor O(1) instead of O(live-sealed)
+    // work that would otherwise grow with a producer-outpaces-drain backlog.
+    // Rotation seals only already-durable predecessors (the
+    // requestSyncBeforeRotation gate), so the sole source of non-durable sealed
+    // segments is a crash-recovery resume; those are covered because the
+    // frontier starts at 0. Maintained entirely under this monitor, in the same
+    // coordinate space as sealedHead (shifted by compaction, reset on clear).
+    private int firstNonDurableSealed;
     // High-water byte offset within the active segment at which we proactively
     // ask the segment manager to provision a spare (if one isn't already
     // installed). Computed once as 3/4 of segment capacity -- leaves the manager
@@ -668,29 +681,23 @@ private static RetainedSegmentMembership newDefaultMembership(
         }
         if (DEFAULT_MEMBERSHIP_MODE == RetainedSegmentMembershipMode.LINEAR) {
             if (observer == null) {
-                return new RetainedSegmentMembership() {
-                    @Override
-                    public boolean contains(MmapSegment segment) {
-                        for (int i = 0, n = chain.size(); i < n; i++) {
-                            if (chain.get(i) == segment) {
-                                return true;
-                            }
-                        }
-                        return false;
-                    }
-                };
-            }
-            return new RetainedSegmentMembership() {
-                @Override
-                public boolean contains(MmapSegment segment) {
+                return segment -> {
                     for (int i = 0, n = chain.size(); i < n; i++) {
-                        observer.onMembershipOperation();
                         if (chain.get(i) == segment) {
                             return true;
                         }
                     }
                     return false;
+                };
+            }
+            return segment -> {
+                for (int i = 0, n = chain.size(); i < n; i++) {
+                    observer.onMembershipOperation();
+                    if (chain.get(i) == segment) {
+                        return true;
+                    }
                 }
+                return false;
             };
         }
 
@@ -699,19 +706,11 @@ public boolean contains(MmapSegment segment) {
             retained.put(chain.get(i), Boolean.TRUE);
         }
         if (observer == null) {
-            return new RetainedSegmentMembership() {
-                @Override
-                public boolean contains(MmapSegment segment) {
-                    return retained.containsKey(segment);
-                }
-            };
+            return retained::containsKey;
         }
-        return new RetainedSegmentMembership() {
-            @Override
-            public boolean contains(MmapSegment segment) {
-                observer.onMembershipOperation();
-                return retained.containsKey(segment);
-            }
+        return segment -> {
+            observer.onMembershipOperation();
+            return retained.containsKey(segment);
         };
     }
 
@@ -1070,6 +1069,46 @@ synchronized void copyLiveSegmentsForSync(ObjList<MmapSegment> target) {
         }
     }
 
+    /**
+     * Copies the live segments that may still need a durability barrier: every
+     * sealed segment from the {@link #firstNonDurableSealed} frontier onward,
+     * plus the active segment. First advances the frontier past any sealed
+     * segments an earlier pass (or rotation's pre-seal barrier) has since made
+     * durable. Used by the periodic sync path in place of
+     * {@link #copyLiveSegmentsForSync}: the proven-durable prefix would
+     * otherwise be re-copied under this monitor and re-scanned every tick as
+     * no-op {@link MmapSegment#syncPublished()} early-returns -- O(live-sealed)
+     * work that grows without bound under a producer-outpaces-drain backlog.
+     * The frontier is a conservative lower bound (it only ever advances past
+     * segments observed durable, and durability never regresses), so this can
+     * never skip a segment that still needs a barrier.
+     */
+    synchronized void copyPendingSyncSegments(ObjList<MmapSegment> target) {
+        target.clear();
+        // Invariant maintained by every mutation site (rotation append, trim's
+        // removeSealedHead, compaction shift, close). A frontier that drifted
+        // above size would silently skip un-fsynced segments, so guard it in
+        // tests; the clamp below keeps production safe if it is ever violated.
+        assert firstNonDurableSealed >= sealedHead && firstNonDurableSealed <= sealedSegments.size()
+                : "durability frontier out of range: firstNonDurableSealed=" + firstNonDurableSealed
+                + " sealedHead=" + sealedHead + " size=" + sealedSegments.size();
+        int i = firstNonDurableSealed;
+        if (i < sealedHead) {
+            i = sealedHead;
+        }
+        int n = sealedSegments.size();
+        while (i < n && sealedSegments.get(i).isPublishedDurable()) {
+            i++;
+        }
+        firstNonDurableSealed = i;
+        for (; i < n; i++) {
+            target.add(sealedSegments.get(i));
+        }
+        if (active != null) {
+            target.add(active);
+        }
+    }
+
     void enablePeriodicSync() {
         periodicSyncEnabled = true;
         syncRequested = true;
@@ -1115,6 +1154,7 @@ public synchronized void close() {
         }
         sealedSegments.clear();
         sealedHead = 0;
+        firstNonDurableSealed = 0;
         for (int i = 0, n = pendingTrims.size(); i < n; i++) {
             pendingTrims.getQuick(i).close();
         }
@@ -1211,7 +1251,6 @@ public synchronized MmapSegment firstTrimmable() {
         return lastSeq <= ackedFsn ? segment : null;
     }
 
-    /** Active segment -- exposed for the I/O thread's "send next batch" path. */
     /**
      * Walks every published frame in the ring (sealed segments plus the active
      * segment) and returns the FSN of the LAST frame whose payload does NOT
@@ -1419,6 +1458,20 @@ public synchronized int getPendingTrimCount() {
         return pendingTrims.size();
     }
 
+    /**
+     * Number of live segments the periodic path would barrier this tick,
+     * advancing the durability frontier exactly as a real tick does. In the
+     * steady state (every sealed segment proven durable) this collapses to 1
+     * -- the active segment -- proving the proven-durable sealed prefix is no
+     * longer copied/scanned under the monitor.
+     */
+    @TestOnly
+    public synchronized int pendingSyncSegmentCountForTest() {
+        ObjList<MmapSegment> scratch = new ObjList<>();
+        copyPendingSyncSegments(scratch);
+        return scratch.size();
+    }
+
     @TestOnly
     public synchronized MmapSegment pinSegmentContainingForTest(long fsn) {
         return pinSegmentContaining(fsn);
@@ -1618,16 +1671,30 @@ private void compactSealedSegments() {
             int liveCount = sealedSegments.size() - sealedHead;
             trimMovedReferences += liveCount;
             sealedSegments.remove(0, sealedHead - 1);
+            // The durable-prefix frontier lives in the same index space as the
+            // entries we just shifted down by sealedHead; move it with them.
+            // The invariant firstNonDurableSealed >= sealedHead keeps this >= 0.
+            firstNonDurableSealed -= sealedHead;
+            if (firstNonDurableSealed < 0) {
+                firstNonDurableSealed = 0;
+            }
             sealedHead = 0;
         }
     }
 
     private void removeSealedHead() {
         sealedSegments.setQuick(sealedHead++, null);
+        // If the removed head WAS the frontier (a non-durable but already-ACKed
+        // recovery-resumed segment can be trimmed before its first barrier),
+        // keep the frontier at or ahead of the live head.
+        if (firstNonDurableSealed < sealedHead) {
+            firstNonDurableSealed = sealedHead;
+        }
         int size = sealedSegments.size();
         if (sealedHead == size) {
             sealedSegments.clear();
             sealedHead = 0;
+            firstNonDurableSealed = 0;
         } else if (sealedHead >= 64 && sealedHead >= size - sealedHead) {
             compactSealedSegments();
         }

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

@@ -305,6 +305,87 @@ public void testSyncPassStopsAtFirstFailureThenRetryCoversAllSegments() throws E
         });
     }
 
+    @Test
+    public void testPeriodicFrontierSkipsSealedPrefixOnceDurable() throws Exception {
+        TestUtils.assertMemoryLeak(() -> {
+            // Recovery resumes a non-durable sealed segment (durableCursor at
+            // the header) plus a non-durable active. The first periodic copy
+            // must offer BOTH -- the sealed one still needs a barrier. Once a
+            // pass has made them durable, the durability frontier advances and
+            // later copies must offer ONLY the active: the proven-durable
+            // sealed prefix is never re-copied/re-scanned under the ring
+            // monitor. This guards the O(1)-steady-state contract that
+            // replaced the old O(live-sealed) copyLiveSegmentsForSync pass.
+            final long intervalNanos = 100L;
+            final long segmentSize = MmapSegment.HEADER_SIZE
+                    + 2 * (MmapSegment.FRAME_HEADER_SIZE + 16);
+            AtomicLong ticks = new AtomicLong();
+            CountingFilesFacade filesFacade = new CountingFilesFacade();
+            String dir = TestUtils.createTmpDir("qdb-periodic-frontier-");
+            long payload = Unsafe.malloc(16, MemoryTag.NATIVE_DEFAULT);
+            SegmentManager manager = null;
+            SegmentRing ring = null;
+            try {
+                // Sealed chain member (full: FSNs 0..1) + active (FSN 2).
+                try (MmapSegment s0 = MmapSegment.create(
+                        filesFacade, dir + "/r0.sfa", 0L, segmentSize)) {
+                    s0.tryAppend(payload, 16);
+                    s0.tryAppend(payload, 16);
+                    s0.msync();
+                }
+                try (MmapSegment s1 = MmapSegment.create(
+                        filesFacade, dir + "/r1.sfa", 2L, segmentSize)) {
+                    s1.tryAppend(payload, 16);
+                    s1.msync();
+                }
+                ring = SegmentRing.openExisting(filesFacade, dir, segmentSize);
+                assertTrue(ring != null);
+                assertEquals(1, ring.getSealedSegments().size());
+
+                manager = new SegmentManager(
+                        segmentSize,
+                        SegmentManager.DEFAULT_POLL_NANOS,
+                        segmentSize * 8L,
+                        filesFacade,
+                        ticks::get);
+                manager.register(ring, dir, null, intervalNanos);
+
+                // Before any successful barrier the recovered sealed segment is
+                // non-durable, so the frontier cannot advance past it: the copy
+                // must offer sealed + active.
+                assertEquals("non-durable recovered sealed segment must be offered for sync",
+                        2, ring.pendingSyncSegmentCountForTest());
+
+                // Run the pass (enablePeriodicSync requested it): both segments
+                // are barriered and become durable.
+                manager.serviceRingForTesting(ring);
+
+                // The sealed segment is still live (nothing ACKed, so no trim)
+                // but now durable, so the frontier skips it: only the active is
+                // offered. count == 1 with a still-present sealed segment proves
+                // the skip.
+                assertEquals("durable sealed segment must remain live (no trim without ACKs)",
+                        1, ring.getSealedSegments().size());
+                assertEquals("durable sealed prefix must be skipped by the periodic copy",
+                        1, ring.pendingSyncSegmentCountForTest());
+                // Idempotent: re-running the frontier advance changes nothing.
+                assertEquals(1, ring.pendingSyncSegmentCountForTest());
+            } finally {
+                if (manager != null && ring != null) {
+                    manager.deregister(ring);
+                }
+                if (ring != null) {
+                    ring.close();
+                }
+                if (manager != null) {
+                    manager.close();
+                }
+                Unsafe.free(payload, 16, MemoryTag.NATIVE_DEFAULT);
+                TestUtils.removeTmpDir(dir);
+            }
+        });
+    }
+
     @Test
     public void testTransientSyncFailureClearsOnNextSuccess() throws Exception {
         TestUtils.assertMemoryLeak(() -> {