fix(ilp): cursor recovery — derive next FSN from on-disk segments

CursorSendEngine's constructor unconditionally created a fresh
sf-initial.sfa at baseSeq=0 even when the SF directory contained
sealed segments from a prior session, restarting the FSN sequence
at 0 and overlapping with FSNs already on disk. ACK translation,
trim, and replay would then operate on overlapping ranges.

The recovery primitive — SegmentRing.openExisting — already exists
and does the right thing (scans *.sfa, sorts by baseSeq, validates
contiguity, picks the highest-baseSeq segment as the new active).
The constructor just never called it. Now it does, in disk mode,
before falling back to the fresh-create path on an empty dir.

Adds a regression test that writes 5 frames to one engine, closes,
reopens against the same dir, and asserts the next append's FSN
continues at 5 instead of restarting at 0.

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/CursorSendEngine.java

@@ -144,25 +144,32 @@ private CursorSendEngine(String sfDir, long segmentSizeBytes, SegmentManager man
         this.ownsManager = ownsManager;
         this.appendDeadlineNanos = appendDeadlineNanos;
 
-        // Create the initial active segment with baseSeq=0. (No on-disk
-        // recovery in PR1 — assumes the directory is empty.) The manager will
-        // immediately notice that the ring needs a hot spare and provision one.
-        MmapSegment initial;
-        String initialPath = null;
-        if (memoryMode) {
-            initial = MmapSegment.createInMemory(0L, segmentSizeBytes);
+        // Disk mode: try to recover any *.sfa files left behind by a prior
+        // session before deciding to start fresh. Without this the engine
+        // would create a new sf-initial.sfa at baseSeq=0, overlapping FSNs
+        // already on disk and corrupting ACK translation, trim, and replay.
+        SegmentRing recovered = memoryMode ? null
+                : SegmentRing.openExisting(sfDir, segmentSizeBytes);
+        if (recovered != null) {
+            this.ring = recovered;
         } else {
-            initialPath = sfDir + "/sf-initial.sfa";
-            initial = MmapSegment.create(initialPath, 0L, segmentSizeBytes);
-        }
-        try {
-            this.ring = new SegmentRing(initial, segmentSizeBytes);
-        } catch (Throwable t) {
-            initial.close();
-            if (initialPath != null) {
-                Files.remove(initialPath);
+            MmapSegment initial;
+            String initialPath = null;
+            if (memoryMode) {
+                initial = MmapSegment.createInMemory(0L, segmentSizeBytes);
+            } else {
+                initialPath = sfDir + "/sf-initial.sfa";
+                initial = MmapSegment.create(initialPath, 0L, segmentSizeBytes);
+            }
+            try {
+                this.ring = new SegmentRing(initial, segmentSizeBytes);
+            } catch (Throwable t) {
+                initial.close();
+                if (initialPath != null) {
+                    Files.remove(initialPath);
+                }
+                throw t;
             }
-            throw t;
         }
 
         if (ownsManager) {

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

@@ -175,6 +175,60 @@ public void testAppendBlockingThrowsOnDeadlineExpiryUnderCap() throws Exception
         }
     }
 
+    @Test
+    public void testRestartIntoNonEmptySfDirContinuesFsnSequence() {
+        // Red regression: restart against a populated SF dir must derive the
+        // new active's baseSeq from the highest sealed segment on disk, not
+        // hardcode 0. Previously CursorSendEngine always created a fresh
+        // sf-initial.sfa at baseSeq=0, so the second session's FSNs collided
+        // with frames the first session had already durably persisted.
+        long segSize = MmapSegment.HEADER_SIZE
+                + 2 * (MmapSegment.FRAME_HEADER_SIZE + 64);
+        int totalFrames = 5;
+        long buf = Unsafe.malloc(64, MemoryTag.NATIVE_DEFAULT);
+        try {
+            // Session 1: write totalFrames, leaving the dir populated with
+            // sealed segments + a (partially-filled) active at the end.
+            try (CursorSendEngine engine = new CursorSendEngine(tmpDir, segSize)) {
+                for (int i = 0; i < totalFrames; i++) {
+                    long fsn = engine.appendBlocking(buf, 64);
+                    assertEquals(i, fsn);
+                }
+                assertEquals(totalFrames - 1, engine.publishedFsn());
+            }
+            // Confirm the dir really has *.sfa files left over — otherwise
+            // the test would pass for the wrong reason (empty dir == no bug).
+            long find = Files.findFirst(tmpDir);
+            assertTrue("findFirst() must succeed on populated tmpDir", find != 0);
+            int sfaCount = 0;
+            try {
+                int rc = 1;
+                while (rc > 0) {
+                    String name = Files.utf8ToString(Files.findName(find));
+                    if (name != null && name.endsWith(".sfa")) sfaCount++;
+                    rc = Files.findNext(find);
+                }
+            } finally {
+                Files.findClose(find);
+            }
+            assertTrue("session 1 must leave .sfa files behind: count=" + sfaCount,
+                    sfaCount >= 1);
+
+            // Session 2: open the same dir. The next FSN must continue from
+            // where session 1 left off, NOT restart at 0. Today this assertion
+            // fails because the engine constructs a fresh ring at baseSeq=0
+            // and ignores the on-disk segments.
+            try (CursorSendEngine engine = new CursorSendEngine(tmpDir, segSize)) {
+                long fsn = engine.appendBlocking(buf, 64);
+                assertEquals("FSN must continue, not restart — overlapping "
+                                + "FSNs would corrupt ACK translation, trim, and replay",
+                        totalFrames, fsn);
+            }
+        } finally {
+            Unsafe.free(buf, 64, MemoryTag.NATIVE_DEFAULT);
+        }
+    }
+
     @Test
     public void testMemoryModeSkipsDirAndStillWorks() {
         // sfDir == null → memory-only ring. No files, no mkdir, no path.