fix(client): batch manifest updates during segment trim

Author: bluestreak01

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

@@ -962,37 +962,63 @@ private boolean serviceRing0(RingEntry e) {
         boolean trimFailed = false;
         Throwable trimFailure = null;
         int unlinked = 0;
+        int batchSize = 0;
         MmapSegment segment = first;
         long coveredAck = durableAck;
-        while (segment != null && unlinked < MAX_TRIMS_PER_RING_PASS) {
+        while (segment != null && batchSize < MAX_TRIMS_PER_RING_PASS) {
             long lastSeq = segment.baseSeq() + segment.frameCount() - 1L;
             if (lastSeq > coveredAck) {
                 break;
             }
-            MmapSegment next = e.ring.nextSealedAfter(segment);
-            String path = segment.path();
+            trimBatch[batchSize++] = segment;
+            segment = e.ring.nextSealedAfter(segment);
+        }
+        if (batchSize > 0) {
             try {
-                // Durably commit the head advance past this segment before
-                // unlinking it. The ring recomputes the successor under its
-                // own monitor -- computing it here from `next` would race
-                // with a concurrent rotation sealing the active, letting the
-                // head leapfrog a still-unacked sealed segment (whose file a
-                // later recovery would then discard as "stale below head").
-                e.ring.advanceManifestHeadPast(segment);
-                segment.close();
-                // A retry after a post-unlink directory-sync failure sees an
-                // already absent path. Treat absence as the prior successful
-                // unlink and retry the batch barrier rather than wedging.
-                if (!filesFacade.remove(path) && filesFacade.exists(path)) {
+                // Durably commit the head advance past the LAST batch member
+                // before unlinking any of them. One commit covers the whole
+                // batch: head values are segment boundaries and the batch is
+                // a contiguous prefix of the sealed chain, so recovery
+                // discards every batch member as "stale below head" whether
+                // the crash lands mid-unlink or after -- byte-identical to a
+                // per-segment commit, minus up to 63 device flushes. The ring
+                // recomputes the successor under its own monitor --
+                // computing it here from the walk above would race with a
+                // concurrent rotation sealing the active, letting the head
+                // leapfrog a still-unacked sealed segment (whose file a later
+                // recovery would then discard as "stale below head").
+                e.ring.advanceManifestHeadPast(trimBatch[batchSize - 1]);
+            } catch (Throwable t) {
+                for (int i = 0; i < batchSize; i++) {
+                    trimBatch[i] = null;
+                }
+                recordTrimFailure(e, TRIM_RETRY_UNLINK, now, t);
+                return false;
+            }
+            while (unlinked < batchSize) {
+                MmapSegment trimming = trimBatch[unlinked];
+                String path = trimming.path();
+                try {
+                    trimming.close();
+                    // A retry after a post-unlink directory-sync failure sees
+                    // an already absent path. Treat absence as the prior
+                    // successful unlink and retry the batch barrier rather
+                    // than wedging. A retry after a mid-batch unlink failure
+                    // re-collects from firstTrimmable(); its head advance
+                    // clamps to the already-committed boundary (no fsync).
+                    if (!filesFacade.remove(path) && filesFacade.exists(path)) {
+                        trimFailed = true;
+                        break;
+                    }
+                    unlinked++;
+                } catch (Throwable t) {
                     trimFailed = true;
+                    trimFailure = t;
                     break;
                 }
-                trimBatch[unlinked++] = segment;
-                segment = next;
-            } catch (Throwable t) {
-                trimFailed = true;
-                trimFailure = t;
-                break;
+            }
+            for (int i = unlinked; i < batchSize; i++) {
+                trimBatch[i] = null;
             }
         }
 

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

@@ -518,13 +518,17 @@ static Recovery recover(
     }
 
     /**
-     * Durably advances the manifest head past {@code trimming} (the sealed
-     * segment the manager is about to unlink). The successor and the current
-     * active are both read under the ring monitor, so a concurrent rotation
-     * (which also mutates the manifest under this monitor) can never make the
-     * head leapfrog a still-live sealed segment: if rotation sealed the old
-     * active after the caller's snapshot, {@code trimming.successor()} now
-     * points at that sealed segment, not at the new active.
+     * Durably advances the manifest head past {@code trimming} (the LAST
+     * sealed segment of the bounded batch the manager is about to unlink).
+     * One durable commit covers every earlier batch member: head values are
+     * segment boundaries and the batch is a contiguous prefix of the sealed
+     * chain, so recovery discards each member as "stale below head"
+     * regardless of how far the unlink loop got. The successor and the
+     * current active are both read under the ring monitor, so a concurrent
+     * rotation (which also mutates the manifest under this monitor) can never
+     * make the head leapfrog a still-live sealed segment: if rotation sealed
+     * the old active after the caller's snapshot, {@code trimming.successor()}
+     * now points at that sealed segment, not at the new active.
      */
     synchronized void advanceManifestHeadPast(MmapSegment trimming) {
         if (manifest == null) {

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

@@ -49,6 +49,10 @@ final class SfManifest implements QuietCloseable {
     private final int fd;
     private final FilesFacade filesFacade;
     private final String path;
+    // Preallocated record scratch for update(): the trim path calls update()
+    // once per batch and must not malloc/free per call. Guarded by this
+    // object's monitor (update() and close() are both synchronized).
+    private final long writeScratch;
     private long activeBase;
     private boolean closed;
     private long generation;
@@ -62,6 +66,7 @@ private SfManifest(FilesFacade filesFacade, String path, int fd,
         this.generation = generation;
         this.headBase = headBase;
         this.activeBase = activeBase;
+        this.writeScratch = Unsafe.malloc(RECORD_SIZE, MemoryTag.NATIVE_DEFAULT);
     }
 
     static SfManifest create(FilesFacade filesFacade, String dir, long headBase, long activeBase) {
@@ -71,11 +76,12 @@ static SfManifest create(FilesFacade filesFacade, String dir, long headBase, lon
             throw new MmapSegmentException("exclusive create failed for SF manifest " + path);
         }
         boolean success = false;
+        SfManifest manifest = null;
         try {
             if (!filesFacade.allocate(fd, FILE_SIZE)) {
                 throw new MmapSegmentException("could not allocate SF manifest " + path);
             }
-            SfManifest manifest = new SfManifest(filesFacade, path, fd, 0, -1, -1);
+            manifest = new SfManifest(filesFacade, path, fd, 0, -1, -1);
             manifest.update(headBase, activeBase);
             if (filesFacade.fsyncDir(dir) != 0) {
                 throw new MmapSegmentException("could not sync SF manifest directory " + dir);
@@ -84,7 +90,13 @@ static SfManifest create(FilesFacade filesFacade, String dir, long headBase, lon
             return manifest;
         } finally {
             if (!success) {
-                filesFacade.close(fd);
+                if (manifest != null) {
+                    // close() frees the constructor-owned scratch buffer as
+                    // well as the fd; closing only the raw fd would leak it.
+                    manifest.close();
+                } else {
+                    filesFacade.close(fd);
+                }
                 filesFacade.remove(path);
             }
         }
@@ -151,9 +163,13 @@ long activeBase() {
     }
 
     @Override
-    public void close() {
+    public synchronized void close() {
+        // Synchronized against update() so the scratch buffer can never be
+        // freed under a concurrent writer; update() checks `closed` inside
+        // the same monitor.
         if (!closed) {
             closed = true;
+            Unsafe.free(writeScratch, RECORD_SIZE, MemoryTag.NATIVE_DEFAULT);
             filesFacade.close(fd);
         }
     }
@@ -202,29 +218,24 @@ synchronized void update(long newHeadBase, long newActiveBase) {
             return;
         }
         long nextGeneration = generation + 1;
-        long buffer = Unsafe.malloc(RECORD_SIZE, MemoryTag.NATIVE_DEFAULT);
-        try {
-            Unsafe.getUnsafe().setMemory(buffer, RECORD_SIZE, (byte) 0);
-            Unsafe.getUnsafe().putInt(buffer, MAGIC);
-            Unsafe.getUnsafe().putInt(buffer + 4, VERSION);
-            Unsafe.getUnsafe().putLong(buffer + 8, nextGeneration);
-            Unsafe.getUnsafe().putLong(buffer + 16, newHeadBase);
-            Unsafe.getUnsafe().putLong(buffer + 24, newActiveBase);
-            int crc = Crc32c.update(Crc32c.INIT, buffer, CRC_OFFSET);
-            Unsafe.getUnsafe().putInt(buffer + CRC_OFFSET, crc);
-            long offset = (nextGeneration & 1L) * RECORD_SIZE;
-            if (filesFacade.write(fd, buffer, RECORD_SIZE, offset) != RECORD_SIZE) {
-                throw new MmapSegmentException("short write updating SF manifest " + path);
-            }
-            if (filesFacade.fsync(fd) != 0) {
-                throw new MmapSegmentException("could not sync SF manifest " + path);
-            }
-            generation = nextGeneration;
-            headBase = newHeadBase;
-            activeBase = newActiveBase;
-        } finally {
-            Unsafe.free(buffer, RECORD_SIZE, MemoryTag.NATIVE_DEFAULT);
+        Unsafe.getUnsafe().setMemory(writeScratch, RECORD_SIZE, (byte) 0);
+        Unsafe.getUnsafe().putInt(writeScratch, MAGIC);
+        Unsafe.getUnsafe().putInt(writeScratch + 4, VERSION);
+        Unsafe.getUnsafe().putLong(writeScratch + 8, nextGeneration);
+        Unsafe.getUnsafe().putLong(writeScratch + 16, newHeadBase);
+        Unsafe.getUnsafe().putLong(writeScratch + 24, newActiveBase);
+        int crc = Crc32c.update(Crc32c.INIT, writeScratch, CRC_OFFSET);
+        Unsafe.getUnsafe().putInt(writeScratch + CRC_OFFSET, crc);
+        long offset = (nextGeneration & 1L) * RECORD_SIZE;
+        if (filesFacade.write(fd, writeScratch, RECORD_SIZE, offset) != RECORD_SIZE) {
+            throw new MmapSegmentException("short write updating SF manifest " + path);
+        }
+        if (filesFacade.fsync(fd) != 0) {
+            throw new MmapSegmentException("could not sync SF manifest " + path);
         }
+        generation = nextGeneration;
+        headBase = newHeadBase;
+        activeBase = newActiveBase;
     }
 
     /**