fix(ilp): plug SF rotate-OOM double-free and on-disk leak

SegmentLog.rotate freed old.pathPtrNative and assigned the new sealed-
path pointer in non-atomic order: free → assign-path → alloc → set
sealed. If allocNativePath OOMed mid-sequence the segment was left in
two simultaneously broken states:

- pathPtrNative still held the freed pointer (the assignment never
  ran). On close() the segments-cleanup loop called freeNativePath
  on it again — a native-heap double-free that crashed the JVM with
  malloc free-list corruption (verified via the new red test on the
  unfixed code: surefire reported "The forked VM terminated without
  properly saying goodbye").
- sealed/lastSeqOnDisk were never set, so trim()'s !s.sealed guard
  silently skipped the segment. The .sfs file on disk was never
  reclaimed within the lifetime of the process.

Fix:
- Set old.pathPtrNative=0 immediately after the free so a subsequent
  OOM cannot leave a stale freed pointer in the field.
- Mark sealed=true / lastSeqOnDisk=lastSeq BEFORE allocating the new
  pointer. After OOM the segment is still classified as sealed so
  trim can reclaim it.
- trim() now handles the recovery case where pathPtrNative is 0 by
  falling back to ff.remove(path) (one-time per trim, acceptable —
  these recovery branches only fire after an OOM, not on the hot
  path).

Test: testRotateOomLeavesSegmentInRecoverableSealedState
(SegmentLogTest). Forces rotation under an injected allocNativePath
OOM, then asserts (a) close() does not double-free, (b) trim() reaps
the orphan .sfs file. On the unfixed code the JVM dies; on the fix
118 SF + adjacent tests 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/SegmentLog.java

@@ -274,9 +274,16 @@ public void trim(long ackedSeq) {
                     ff.close(s.fd);
                     s.fd = -1;
                 }
-                ff.remove(s.pathPtrNative);
-                ff.freeNativePath(s.pathPtrNative);
-                s.pathPtrNative = 0;
+                if (s.pathPtrNative != 0) {
+                    ff.remove(s.pathPtrNative);
+                    ff.freeNativePath(s.pathPtrNative);
+                    s.pathPtrNative = 0;
+                } else {
+                    // Recovery case: rotate's allocNativePath OOMed and left
+                    // pathPtrNative=0. Fall back to the String form, which
+                    // does its own one-shot encode/free internally.
+                    ff.remove(s.path);
+                }
                 bytesOnDiskCache -= s.writePos;
             } else {
                 segments.setQuick(writeIdx++, s);
@@ -561,12 +568,21 @@ private void rotate() {
         if (ff.rename(old.path, sealedPath) != 0) {
             throw new SfException("failed to seal segment by rename " + old.path + " -> " + sealedPath);
         }
-        // Path changed — free old native ptr and re-encode for the sealed name.
+        // Filesystem is now in the sealed state. Update bookkeeping to match
+        // BEFORE re-encoding the path pointer; if allocNativePath OOMs:
+        //   - the stale freed pointer must not be left in the field, or
+        //     close() walks segments and calls freeNativePath on it again
+        //     → native double-free.
+        //   - sealed/lastSeqOnDisk must already be set, or trim never sees
+        //     this segment (the !s.sealed guard skips it) → permanent
+        //     on-disk leak that survives until the next process restart.
+        // trim() handles pathPtrNative==0 by falling back to ff.remove(path).
         ff.freeNativePath(old.pathPtrNative);
+        old.pathPtrNative = 0;
         old.path = sealedPath;
-        old.pathPtrNative = ff.allocNativePath(sealedPath);
         old.sealed = true;
         old.lastSeqOnDisk = lastSeq;
+        old.pathPtrNative = ff.allocNativePath(sealedPath);
         createActive(lastSeq + 1);
     }
 

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

@@ -637,6 +637,109 @@ public void testCreateActiveDoesNotLeakFdOnAllocNativePathOom() throws Exception
         });
     }
 
+    /**
+     * Regression test for {@code rotate}'s mid-reseal OOM window.
+     * <p>
+     * Production order at lines 564-570 (pre-fix):
+     * <pre>
+     *   ff.freeNativePath(old.pathPtrNative);                  // ptr freed
+     *   old.path = sealedPath;
+     *   old.pathPtrNative = ff.allocNativePath(sealedPath);    // CAN throw OOM
+     *   old.sealed = true;
+     *   old.lastSeqOnDisk = lastSeq;
+     * </pre>
+     * If {@code allocNativePath} throws after the freed pointer is left in
+     * the field and before {@code sealed/lastSeqOnDisk} are set:
+     * <ul>
+     *   <li><b>native double-free on close:</b> {@code SegmentLog.close()}
+     *       walks {@code segments} and calls {@code freeNativePath} on the
+     *       stale freed pointer.</li>
+     *   <li><b>permanent on-disk leak:</b> {@code trim()}'s {@code !s.sealed}
+     *       guard skips the segment, so the {@code .sfs} file on disk is
+     *       never reclaimed within the lifetime of this process. Even after
+     *       restart it would re-replay forever (no ACK ever advances past
+     *       its lastSeq because the in-memory state lost it).</li>
+     * </ul>
+     * <p>
+     * The fix sets {@code pathPtrNative=0} immediately after the free and
+     * marks {@code sealed=true; lastSeqOnDisk=lastSeq} BEFORE allocating
+     * the new pointer. {@code trim()} falls back to {@code ff.remove(path)}
+     * when {@code pathPtrNative} is 0.
+     */
+    @Test
+    public void testRotateOomLeavesSegmentInRecoverableSealedState() throws Exception {
+        TestUtils.assertMemoryLeak(() -> {
+            FdTrackingFacade tracker = new FdTrackingFacade();
+            // maxBytes = HEADER_SIZE + FRAME_HEADER_SIZE + 16 = 48; first
+            // append fits the active segment exactly, the second forces
+            // rotation.
+            final long maxBytes = 48;
+            final int payloadSize = 16;
+
+            long buf = Unsafe.malloc(payloadSize, MemoryTag.NATIVE_DEFAULT);
+            try {
+                for (int i = 0; i < payloadSize; i++) {
+                    Unsafe.getUnsafe().putByte(buf + i, (byte) i);
+                }
+
+                SegmentLog log = SegmentLog.open(tmpDir, tracker, maxBytes, 1024, false);
+                try {
+                    long s0 = log.append(buf, payloadSize);
+                    assertEquals(0L, s0);
+
+                    // Arm the OOM at the rotate's allocNativePath(sealedPath).
+                    tracker.failNextSealedAllocNativePath = true;
+                    try {
+                        log.append(buf, payloadSize);
+                        fail("expected OOM during rotate's allocNativePath(sealedPath)");
+                    } catch (Throwable expected) {
+                        String msg = expected.getMessage() == null ? "" : expected.getMessage();
+                        String causeMsg = expected.getCause() == null
+                                || expected.getCause().getMessage() == null
+                                ? "" : expected.getCause().getMessage();
+                        assertTrue("wrong failure: " + expected,
+                                msg.contains("simulated") || msg.contains("OOM")
+                                        || causeMsg.contains("simulated") || causeMsg.contains("OOM"));
+                    }
+
+                    // The segment is sealed on disk and must be classified
+                    // as sealed in memory so trim() can reclaim it. Drop
+                    // every acked seq up to and including the (now-sealed)
+                    // segment's lastSeq, then assert the file is gone.
+                    log.trim(0);
+                } finally {
+                    // close() walks the segments list and frees pathPtrNative
+                    // for each. Under the bug the rotated segment's stale
+                    // freed pointer would be passed to freeNativePath again
+                    // → native double-free. The fix sets pathPtrNative=0
+                    // after the original free so close() skips it.
+                    log.close();
+                }
+
+                // No .sfs file should remain after trim().
+                long find = Files.findFirst(tmpDir);
+                if (find != 0) {
+                    try {
+                        int rc = 1;
+                        while (rc > 0) {
+                            String name = Files.utf8ToString(Files.findName(find));
+                            if (name != null && name.endsWith(".sfs")) {
+                                fail("sealed .sfs file leaked after trim: " + name
+                                        + " — rotate's mid-OOM left the segment unsealed in "
+                                        + "memory so trim's !s.sealed guard skipped it");
+                            }
+                            rc = Files.findNext(find);
+                        }
+                    } finally {
+                        Files.findClose(find);
+                    }
+                }
+            } finally {
+                Unsafe.free(buf, payloadSize, MemoryTag.NATIVE_DEFAULT);
+            }
+        });
+    }
+
     @Test
     public void testCreateActiveDoesNotLeakFdOnFsyncFailure() throws Exception {
         TestUtils.assertMemoryLeak(() -> {
@@ -1024,15 +1127,26 @@ private static class FdTrackingFacade implements FilesFacade {
         // ACTIVE_SUFFIX. Simulates an OOM at the exact moment between
         // openCleanRW and the try-block in createActive. Auto-reset.
         volatile boolean failNextActiveAllocNativePath;
+        // Set true to fault the NEXT allocNativePath whose path ends in
+        // SEALED_SUFFIX. Simulates an OOM in the rotate-then-reseal path
+        // after the file rename succeeded but before the new pointer is
+        // installed. Auto-reset.
+        volatile boolean failNextSealedAllocNativePath;
 
         @Override
         public long allocNativePath(String path) {
-            // ".sfa" is SegmentLog.ACTIVE_SUFFIX (package-private, hardcoded here).
+            // ".sfa" / ".sfs" are SegmentLog.{ACTIVE,SEALED}_SUFFIX
+            // (package-private, hardcoded here).
             if (failNextActiveAllocNativePath && path.endsWith(".sfa")) {
                 failNextActiveAllocNativePath = false;
                 throw CairoException.nonCritical()
                         .put("simulated OOM in allocNativePath: ").put(path);
             }
+            if (failNextSealedAllocNativePath && path.endsWith(".sfs")) {
+                failNextSealedAllocNativePath = false;
+                throw CairoException.nonCritical()
+                        .put("simulated OOM in allocNativePath: ").put(path);
+            }
             return FilesFacade.INSTANCE.allocNativePath(path);
         }