fix(ilp): recovery replays sealed segments from baseSeq, not active

Three correlated bugs that together orphaned data on recovery:

1. CursorWebSocketSendLoop.start() began at engine.activeSegment(),
   skipping every sealed segment on disk. After a crash + restart with
   multiple unacked segments, only the active's tail would replay; all
   sealed-segment data sat orphaned. Fixed by positioning at
   engine.ackedFsn() + 1 (same as swapClient does on reconnect) — the
   cursor walks sealed segments oldest-first and falls through to active
   only when sealed is exhausted. Existing replay test only covered
   single-batch replay (first batch always carries full schemas), so
   the gap wasn't caught.

2. CursorSendEngine recovery left ackedFsn = -1 even when earlier
   segments had been trimmed before the crash (lowestBaseSeq > 0). With
   ackedFsn -1, positionCursorAt(0) would land before any segment exists
   and fall through to active.publishedOffset() — same orphan symptom.
   Fixed by seeding ackedFsn = lowestBaseSeq - 1 on recovery; everything
   trimmed must have been acked, so this is a sound lower bound.

3. SegmentManager.fileGeneration started at 0 even when the slot dir
   already contained sf-0000000000000000.sfa from a prior session.
   Manager would then mint its first hot spare at that name —
   openCleanRW truncates the file, scrambling the in-flight mmap of the
   active segment under the I/O loop. Spec called for this fix at
   line 93 ("seeds fileGeneration to max(existing) + 1"); now done in
   register() by scanning the slot dir for sf-<hex>.sfa files.

Test: RecoveryReplayTest writes 50 multi-segment-spanning rows against a
silent server (no acks), closes fast, then opens a fresh sender against
an ack server pointed at the same slot. Asserts all 50 distinct row
values reach the new server. Without the start() fix, only the active
segment's frames replay (subset). Without the fileGeneration fix, the
in-flight mmap gets clobbered and the cursor walks zero-padded garbage.

Adds getTotalFramesSent / getTotalAcks accessors on QwpWebSocketSender
(used during diagnosis; useful in their own right).

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

Author: bluestreak01

core/src/main/java/io/questdb/client/cutlass/qwp/client/QwpWebSocketSender.java

@@ -973,6 +973,18 @@ public long getTotalReconnectsSucceeded() {
         return l == null ? 0L : l.getTotalReconnects();
     }
 
+    /** Total binary frames the cursor I/O loop has issued to the wire. */
+    public long getTotalFramesSent() {
+        CursorWebSocketSendLoop l = cursorSendLoop;
+        return l == null ? 0L : l.getTotalFramesSent();
+    }
+
+    /** Total binary frames whose ACKs have been received and applied. */
+    public long getTotalAcks() {
+        CursorWebSocketSendLoop l = cursorSendLoop;
+        return l == null ? 0L : l.getTotalAcks();
+    }
+
     /**
      * Frames re-sent on the post-reconnect catch-up window — i.e. frames
      * whose FSN was already on the wire before the drop. Useful for

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

@@ -166,6 +166,22 @@ private CursorSendEngine(String sfDir, long segmentSizeBytes, SegmentManager man
             this.recoveredFromDisk = recovered != null;
             if (recovered != null) {
                 this.ring = recovered;
+                // Seed ackedFsn to one below the lowest segment's baseSeq.
+                // We don't know what was actually acked before the prior
+                // session crashed, but anything trimmed off the ring's
+                // bottom must have been acked (trim is ack-driven). Without
+                // this seed, ackedFsn stays at -1 and the I/O loop's
+                // start-time positioning would walk to FSN 0 — which may
+                // not exist on disk if earlier segments have been trimmed,
+                // causing it to fall through to the active segment's tip
+                // and skip the unacked sealed segments entirely.
+                MmapSegment first = recovered.firstSealed();
+                long lowestBase = first != null
+                        ? first.baseSeq()
+                        : recovered.getActive().baseSeq();
+                if (lowestBase > 0) {
+                    recovered.acknowledge(lowestBase - 1);
+                }
             } else {
                 MmapSegment initial;
                 String initialPath = null;

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

@@ -269,12 +269,33 @@ public synchronized void start() {
             throw new IllegalStateException("already started");
         }
         running = true;
-        sendingSegment = engine.activeSegment();
+        // Position the cursor at the first unsent FSN before spinning the
+        // I/O thread. For a fresh sender, ackedFsn=-1 → start at FSN 0,
+        // which lands on the (empty) initial active — same as the prior
+        // hardcoded "sendingSegment = engine.activeSegment()". For a
+        // recovered sender with sealed segments holding unsent data, this
+        // walks back to the lowest unacked frame so sealed-segment data
+        // actually reaches the wire — without it, start() would skip
+        // straight to the active and orphan everything in sealed.
+        positionCursorForStart();
         ioThread = new Thread(this::ioLoop, "qdb-cursor-ws-io");
         ioThread.setDaemon(true);
         ioThread.start();
     }
 
+    /**
+     * Sets {@code fsnAtZero}, {@code nextWireSeq}, and the cursor
+     * (sendingSegment + sendOffset) to the first unsent FSN. Visible for
+     * tests so they can assert correct positioning without spinning a
+     * real I/O thread + WebSocket.
+     */
+    void positionCursorForStart() {
+        long replayStart = engine.ackedFsn() + 1L;
+        this.fsnAtZero = replayStart;
+        this.nextWireSeq = 0L;
+        positionCursorAt(replayStart);
+    }
+
     /**
      * Walks to the next segment when the current one is sealed and fully
      * drained. Returns the next segment to consume (newer sealed if available,

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

@@ -169,10 +169,57 @@ public void deregister(SegmentRing ring) {
     public void register(SegmentRing ring, String dir) {
         synchronized (lock) {
             rings.add(new RingEntry(ring, dir));
+            // Skip the file-generation counter past whatever's already on
+            // disk in this slot. Without this, on recovery the manager
+            // would mint a new spare at sf-0000000000000000.sfa — and
+            // openCleanRW would truncate the user's existing active file
+            // out from under the I/O loop, scrambling the in-flight mmap.
+            // Memory-mode rings have no dir; nothing to scan there.
+            if (dir != null) {
+                long minNext = scanMaxGeneration(dir) + 1L;
+                while (true) {
+                    long cur = fileGeneration.get();
+                    if (cur >= minNext) break;
+                    if (fileGeneration.compareAndSet(cur, minNext)) break;
+                }
+            }
         }
         ring.setManagerWakeup(this::wakeWorker);
     }
 
+    /**
+     * Returns the highest hex-encoded generation across {@code sf-<gen>.sfa}
+     * files in {@code dir}, or {@code -1} if none exist. Skips files that
+     * don't match the pattern (e.g. the legacy {@code sf-initial.sfa}).
+     */
+    private static long scanMaxGeneration(String dir) {
+        long max = -1L;
+        if (!Files.exists(dir)) return max;
+        long find = Files.findFirst(dir);
+        if (find == 0) return max;
+        try {
+            int rc = 1;
+            while (rc > 0) {
+                String name = Files.utf8ToString(Files.findName(find));
+                rc = Files.findNext(find);
+                if (name == null || !name.startsWith("sf-") || !name.endsWith(".sfa")) {
+                    continue;
+                }
+                String hex = name.substring(3, name.length() - 4);
+                if (hex.length() != 16) continue;
+                try {
+                    long gen = Long.parseUnsignedLong(hex, 16);
+                    if (gen > max) max = gen;
+                } catch (NumberFormatException ignored) {
+                    // sf-initial.sfa or non-hex — skip
+                }
+            }
+        } finally {
+            Files.findClose(find);
+        }
+        return max;
+    }
+
     /**
      * Unparks the worker thread out of its poll-park so it processes
      * registered rings on the very next loop iteration. Cheap — a single

core/src/test/java/io/questdb/client/test/cutlass/qwp/client/RecoveryReplayTest.java

@@ -0,0 +1,260 @@
+/*+*****************************************************************************
+ *     ___                  _   ____  ____
+ *    / _ \ _   _  ___  ___| |_|  _ \| __ )
+ *   | | | | | | |/ _ \/ __| __| | | |  _ \
+ *   | |_| | |_| |  __/\__ \ |_| |_| | |_) |
+ *    \__\_\\__,_|\___||___/\__|____/|____/
+ *
+ *  Copyright (c) 2014-2019 Appsicle
+ *  Copyright (c) 2019-2026 QuestDB
+ *
+ *  Licensed under the Apache License, Version 2.0 (the "License");
+ *  you may not use this file except in compliance with the License.
+ *  You may obtain a copy of the License at
+ *
+ *  http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ *
+ ******************************************************************************/
+
+package io.questdb.client.test.cutlass.qwp.client;
+
+import io.questdb.client.Sender;
+import io.questdb.client.std.Files;
+import io.questdb.client.test.cutlass.qwp.websocket.TestWebSocketServer;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.nio.file.Paths;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+
+/**
+ * Pin-down for recovery replay across sender restarts.
+ * <p>
+ * Previously {@code CursorWebSocketSendLoop.start()} began at the active
+ * segment, skipping every sealed segment on disk. After a crash + restart
+ * with multiple segments holding unacked data, the foreground sender
+ * would orphan everything in sealed and only ship the active's tail.
+ * <p>
+ * Today {@code start()} positions at {@code engine.ackedFsn() + 1} —
+ * walking sealed segments oldest-first — and the engine constructor
+ * seeds {@code ackedFsn} to {@code lowestBaseSeq - 1} on recovery so the
+ * positioning lands on the right segment even if earlier ones were
+ * trimmed before the crash.
+ */
+public class RecoveryReplayTest {
+
+    private static final int TEST_PORT = 19_100 + (int) (System.nanoTime() % 100);
+    private String sfDir;
+
+    @Before
+    public void setUp() {
+        sfDir = Paths.get(System.getProperty("java.io.tmpdir"),
+                "qdb-recov-replay-" + System.nanoTime()).toString();
+    }
+
+    @After
+    public void tearDown() {
+        if (sfDir != null) rmDirRec(sfDir);
+    }
+
+    @Test
+    public void testRestartReplaysSealedSegmentsAgainstFreshServer() throws Exception {
+        // Phase 1: silent server, sender 1 writes enough to rotate at
+        // least once, closes fast (no drain). Slot ends up with sealed +
+        // active segments holding unacked data.
+        int port1 = TEST_PORT + 1;
+        try (TestWebSocketServer silent = new TestWebSocketServer(port1, new SilentHandler())) {
+            silent.start();
+            Assert.assertTrue(silent.awaitStart(5, TimeUnit.SECONDS));
+
+            // Use a tight segment cap and pad each row with a sizable
+            // payload so 50 batches genuinely span multiple segments.
+            // Without rotation there'd be no sealed segments and the
+            // start-position bug couldn't manifest — defeating the test.
+            String pad = repeat("x", 64);
+            String cfg1 = "ws::addr=localhost:" + port1
+                    + ";sf_dir=" + sfDir
+                    + ";sf_max_bytes=4096"
+                    + ";close_flush_timeout_millis=0;";
+            try (Sender s1 = Sender.fromConfig(cfg1)) {
+                for (int i = 0; i < 50; i++) {
+                    s1.table("foo").stringColumn("p", pad).longColumn("v", (long) i).atNow();
+                    s1.flush();
+                }
+            }
+        }
+
+        // Sanity: the slot must hold at least one sealed segment (one
+        // that's been rotated out of active and closed). We verify by
+        // checking publishedFsn jumps across the active segment's base
+        // seq when re-opened — i.e. there's data in a segment older than
+        // the active.
+        int populatedCount = countPopulatedSegmentFiles(sfDir + "/default");
+        Assert.assertTrue("expected multi-segment slot with data, got "
+                        + populatedCount + " populated .sfa files",
+                populatedCount >= 2);
+
+        // Phase 2: fresh server that ACKs every binary frame. Sender 2
+        // opens the same slot. The bug-fix expectation: every frame
+        // sender 1 wrote (50 of them) reaches the new server. Without
+        // the fix, the sender would only ship the active segment's data
+        // (≪ 50) and orphan the sealed segments forever.
+        int port2 = port1 + 50;
+        AckHandler ack = new AckHandler();
+        try (TestWebSocketServer good = new TestWebSocketServer(port2, ack)) {
+            good.start();
+            Assert.assertTrue(good.awaitStart(5, TimeUnit.SECONDS));
+
+            String cfg2 = "ws::addr=localhost:" + port2
+                    + ";sf_dir=" + sfDir + ";";
+            try (Sender s2 = Sender.fromConfig(cfg2)) {
+                // No new appends — purely replay.
+                long deadline = System.currentTimeMillis() + 5_000;
+                while (System.currentTimeMillis() < deadline
+                        && ack.distinctPayloadHashes.size() < 50) {
+                    Thread.sleep(20);
+                }
+            }
+            // Each row carries a unique long, so every frame's bytes are
+            // distinct. With the start-position fix we expect all 50 of
+            // sender 1's rows to reach server 2; without the fix the cursor
+            // would skip straight to the active segment and orphan
+            // everything in sealed.
+            Assert.assertEquals(
+                    "every distinct row written by sender 1 must replay through to server 2",
+                    50, ack.distinctPayloadHashes.size());
+        }
+    }
+
+    private static int countSegmentFiles(String dir) {
+        if (!Files.exists(dir)) return 0;
+        long find = Files.findFirst(dir);
+        if (find == 0) return 0;
+        int n = 0;
+        try {
+            int rc = 1;
+            while (rc > 0) {
+                String name = Files.utf8ToString(Files.findName(find));
+                if (name != null && name.endsWith(".sfa")) n++;
+                rc = Files.findNext(find);
+            }
+        } finally {
+            Files.findClose(find);
+        }
+        return n;
+    }
+
+    /**
+     * Counts only segment files that actually carry frames — opens each
+     * .sfa via the cursor's MmapSegment recovery path and excludes the
+     * empty hot-spares the segment manager pre-allocates. Without this
+     * filter, the multi-segment sanity check could pass for the wrong
+     * reason on a deployment that's only used a single segment.
+     */
+    private static int countPopulatedSegmentFiles(String dir) {
+        if (!Files.exists(dir)) return 0;
+        long find = Files.findFirst(dir);
+        if (find == 0) return 0;
+        int n = 0;
+        try {
+            int rc = 1;
+            while (rc > 0) {
+                String name = Files.utf8ToString(Files.findName(find));
+                if (name != null && name.endsWith(".sfa")) {
+                    try {
+                        io.questdb.client.cutlass.qwp.client.sf.cursor.MmapSegment seg =
+                                io.questdb.client.cutlass.qwp.client.sf.cursor.MmapSegment
+                                        .openExisting(dir + "/" + name);
+                        try {
+                            if (seg.frameCount() > 0) n++;
+                        } finally {
+                            seg.close();
+                        }
+                    } catch (Throwable ignored) {
+                        // best-effort
+                    }
+                }
+                rc = Files.findNext(find);
+            }
+        } finally {
+            Files.findClose(find);
+        }
+        return n;
+    }
+
+    private static String repeat(String c, int n) {
+        StringBuilder sb = new StringBuilder(n);
+        for (int i = 0; i < n; i++) sb.append(c);
+        return sb.toString();
+    }
+
+    private static void rmDirRec(String dir) {
+        if (!Files.exists(dir)) return;
+        long find = Files.findFirst(dir);
+        if (find != 0) {
+            try {
+                int rc = 1;
+                while (rc > 0) {
+                    String name = Files.utf8ToString(Files.findName(find));
+                    if (name != null && !".".equals(name) && !"..".equals(name)) {
+                        String child = dir + "/" + name;
+                        if (!Files.remove(child)) rmDirRec(child);
+                    }
+                    rc = Files.findNext(find);
+                }
+            } finally {
+                Files.findClose(find);
+            }
+        }
+        Files.remove(dir);
+    }
+
+    /** Receives binary frames but never acks. Sender drops them on close. */
+    private static class SilentHandler implements TestWebSocketServer.WebSocketServerHandler {
+        @Override
+        public void onBinaryMessage(TestWebSocketServer.ClientHandler client, byte[] data) {
+            // intentionally empty
+        }
+    }
+
+    /** Acks every binary frame and tracks distinct payloads. */
+    private static class AckHandler implements TestWebSocketServer.WebSocketServerHandler {
+        // Distinct *payload bytes* — each row carries a unique long value
+        // so every frame's bytes differ. Counts unique frames received,
+        // independent of any amplification (re-sends, fragmentation).
+        final java.util.Set<String> distinctPayloadHashes =
+                java.util.Collections.synchronizedSet(new java.util.HashSet<>());
+        private final AtomicLong nextSeq = new AtomicLong(0);
+
+        @Override
+        public void onBinaryMessage(TestWebSocketServer.ClientHandler client, byte[] data) {
+            distinctPayloadHashes.add(java.util.Arrays.toString(data));
+            try {
+                client.sendBinary(buildAck(nextSeq.getAndIncrement()));
+            } catch (IOException e) {
+                throw new RuntimeException(e);
+            }
+        }
+
+        static byte[] buildAck(long seq) {
+            byte[] buf = new byte[1 + 8 + 2];
+            ByteBuffer bb = ByteBuffer.wrap(buf).order(ByteOrder.LITTLE_ENDIAN);
+            bb.put((byte) 0x00);
+            bb.putLong(seq);
+            bb.putShort((short) 0);
+            return buf;
+        }
+    }
+}