fix(ilp): trim acked frames from active SF segment

Pre-fix behaviour: SegmentLog.trim only deleted sealed segments. Frames
that the server had acknowledged but lived in the still-open active
segment stayed on disk until the next natural rotation. On restart the
new sender replayed those frames and the public Sender.storeAndForward
contract — "trimmed when the server acknowledges it" — was load-bearing
on server-side seqTxn dedup to avoid duplicate rows. Worst case at the
default 64 MiB segment size: ~640 acked batches re-shipped per restart.

Fix: when every frame in the active segment has been acked, force-rotate
the active (sealing the file) and immediately remove the just-sealed
segment. nextSeq is preserved across the auto-rotate so subsequent
appends keep monotonic FSNs. The only safe-guard is the rotate-OOM
recovery state from the M2 fix: when active.sealed is already true,
the sealed pass above has already trimmed the file and force-rotate
is skipped.

Tests:
- testTrimRotatesAndDropsFullyAckedActiveSegment (SegmentLogTest):
  unit-level proof that trim with full coverage drops the active
  contents to a fresh empty segment, with nextSeq preserved.
- testTrimPartialAckOfActiveLeavesItIntact (SegmentLogTest, replaces
  testTrimNeverDeletesActive): proves partial ACKs do not seal a
  segment that still contains unacked data.
- testRestartAfterAckedBatchesReplaysNothing (SfIntegrationTest):
  end-to-end. Send 5 batches, wait for trim, close, reopen with a
  fresh sender, send one more, assert server saw exactly 6 frames
  (5 originals + 1 new, no replays).
- testCapturedBytesMatchWireBytes (SfIntegrationTest): updated to use
  a non-acking handler so the test thread's log.replay() doesn't race
  the I/O thread's trim.
- testAutoReconnectAndReplay (SfIntegrationTest): expected frame
  count drops from 5 to 4 (msg1 trimmed before reconnect, no replay).
- testMultiTableSurvivesReconnect (SfIntegrationTest): expected frame
  count drops from 6 to 5 (alpha-1 trimmed before reconnect).

Public API:
- Sender.storeAndForwardDir Javadoc rewritten to honestly describe the
  new contract: acked batches are reclaimed in real time; only batches
  whose ACK had not been received before sender shutdown are replayed
  on the next sender against the same directory.

Full suite: 1975 tests pass, zero regressions.

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

Author: bluestreak01

core/src/main/java/io/questdb/client/Sender.java

@@ -1569,9 +1569,22 @@ public LineSenderBuilder storeAndForward(boolean enabled) {
         /**
          * Set the store-and-forward directory. Has effect only when SF is also
          * enabled via {@link #storeAndForward(boolean)} (or {@code store_and_forward=on}
-         * in the connect string). Every batch is persisted before it leaves the
-         * wire and trimmed when the server acknowledges it; on restart the sender
-         * replays whatever is on disk. WebSocket transport only.
+         * in the connect string).
+         * <p>
+         * Every batch is persisted to disk before it leaves the wire and is
+         * reclaimed as soon as the server acknowledges it. On restart the
+         * sender replays only batches whose acknowledgement had not been
+         * received before the previous sender shut down — typically the last
+         * in-flight batches at close time. Acknowledged batches are not
+         * replayed: their disk space is freed during normal operation by an
+         * automatic per-frame trim that force-rotates the active segment
+         * once every frame in it has been acknowledged.
+         * <p>
+         * Note that {@link io.questdb.client.cutlass.qwp.client.QwpWebSocketSender#close()}
+         * under SF returns once data is on disk, not on server-ack, so a
+         * sender closed immediately after a flush may still have unacked
+         * batches in flight; those will be replayed by the next sender
+         * against the same directory. WebSocket transport only.
          * <p>
          * The sender takes ownership of the underlying SegmentLog and closes it
          * when the sender itself is closed.

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

@@ -256,12 +256,49 @@ public void replay(FrameVisitor visitor) {
     }
 
     /**
-     * Delete every sealed segment whose lastSeq is <= ackedSeq. The active
-     * segment is never trimmed, even if all of its frames are acked — it is only
-     * deleted when sealed by a rotation.
+     * Reclaim disk space for every frame whose seq is <= ackedSeq.
+     * <p>
+     * Sealed segments whose {@code lastSeq <= ackedSeq} are deleted. If the
+     * current active segment also has all of its frames acked (i.e. its
+     * highest assigned seq &lt;= ackedSeq), it is force-rotated and the
+     * just-sealed file is immediately removed. {@code nextSeq} is preserved
+     * across the auto-rotate so future appends keep monotonic FSNs.
+     * <p>
+     * The force-rotate is what makes "trimmed when the server acknowledges
+     * it" honest in the public API: a quiet sender whose batches all
+     * acknowledge keeps disk at exactly one empty active segment, and on
+     * restart no acked frames are replayed.
      */
     public void trim(long ackedSeq) {
         ensureOpen();
+        trimSealedSegments(ackedSeq);
+
+        // Force-rotate the active segment when every frame in it has been
+        // acked. The just-sealed segment is then removed by a second pass
+        // of trimSealedSegments. Cost is one extra rotation per fully-acked
+        // burst (typically once per server cumulative ACK), which on a
+        // low-rate sender is amortised by the natural-rotation cost it
+        // displaces — the active will rotate anyway eventually.
+        //
+        // The {@code !active.sealed} guard handles the rotate-OOM recovery
+        // state from the M2 fix: after an OOM mid-rename, {@code active}
+        // points at the now-sealed segment with fd=-1 and pathPtrNative=0;
+        // attempting to rotate it again would fail in fsync. The sealed
+        // pass above already trimmed the file, so we just skip here.
+        //
+        // If rotate fails (e.g. fsync EIO), the SfException propagates to
+        // the caller. ResponseHandler.onBinaryMessage runs trim() inline
+        // with ACK processing, so a thrown SfException there will surface
+        // as a connection-level error and the sender goes terminal — the
+        // correct response to a broken disk.
+        if (active != null && !active.sealed && active.frameCount > 0
+                && active.baseSeq + active.frameCount - 1 <= ackedSeq) {
+            rotate();
+            trimSealedSegments(ackedSeq);
+        }
+    }
+
+    private void trimSealedSegments(long ackedSeq) {
         int writeIdx = 0;
         for (int i = 0, n = segments.size(); i < n; i++) {
             Segment s = segments.getQuick(i);

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

@@ -246,8 +246,14 @@ public void testTrimDeletesSealedFullyAcked() throws Exception {
         });
     }
 
+    /**
+     * When ACK covers some-but-not-all of the active segment's frames, the
+     * active segment must remain on disk (force-rotate only fires when
+     * every frame is acked). Without this guard a partially-acked active
+     * would be sealed and the unacked frames would be silently lost.
+     */
     @Test
-    public void testTrimNeverDeletesActive() throws Exception {
+    public void testTrimPartialAckOfActiveLeavesItIntact() throws Exception {
         TestUtils.assertMemoryLeak(() -> {
             byte[] payload = "x".getBytes();
             try (SegmentLog log = SegmentLog.open(tmpDir, 1L << 20)) {
@@ -259,15 +265,77 @@ public void testTrimNeverDeletesActive() throws Exception {
                     Unsafe.free(buf, payload.length, MemoryTag.NATIVE_DEFAULT);
                 }
                 log.fsync();
-                // ack way past everything; active is unsealed so must remain.
-                log.trim(Long.MAX_VALUE / 2);
+                // Ack only the first frame; second is still in flight. The
+                // active must NOT be force-rotated yet — that would seal a
+                // segment containing un-acked data.
+                log.trim(0);
                 assertEquals(1, log.segmentCount());
                 int[] count = {0};
                 log.replay((seq, addr, len) -> {
                     count[0]++;
                     return true;
                 });
-                assertEquals(2, count[0]);
+                assertEquals("both frames must still be on disk", 2, count[0]);
+            }
+        });
+    }
+
+    /**
+     * When ACK covers every frame in the active segment, the active is
+     * force-rotated and the just-sealed segment removed. nextSeq is
+     * preserved across the auto-rotate so subsequent appends keep
+     * monotonic FSNs. After reopen, replay yields zero frames — this is
+     * what makes "trimmed when the server acknowledges it" honest in the
+     * public Sender API.
+     */
+    @Test
+    public void testTrimRotatesAndDropsFullyAckedActiveSegment() throws Exception {
+        TestUtils.assertMemoryLeak(() -> {
+            byte[] payload = "x".getBytes();
+            try (SegmentLog log = SegmentLog.open(tmpDir, 1L << 20)) {
+                long buf = alloc(payload);
+                try {
+                    for (int i = 0; i < 5; i++) {
+                        log.append(buf, payload.length);
+                    }
+                } finally {
+                    Unsafe.free(buf, payload.length, MemoryTag.NATIVE_DEFAULT);
+                }
+                log.fsync();
+
+                long preTrimBytes = log.bytesOnDisk();
+                assertTrue("data must be on disk before trim",
+                        preTrimBytes > SegmentLog.HEADER_SIZE);
+                assertEquals(5L, log.nextSeq());
+
+                // Ack every frame; force-rotate kicks in, sealed segment
+                // removed in the same trim() call.
+                log.trim(4);
+
+                assertEquals("a fresh empty active must remain", 1, log.segmentCount());
+                assertEquals("nextSeq must survive the auto-rotate", 5L, log.nextSeq());
+                assertEquals("oldestSeq must report empty (no frames)", -1L, log.oldestSeq());
+                assertEquals("only the new active's header should be on disk",
+                        (long) SegmentLog.HEADER_SIZE, log.bytesOnDisk());
+                int[] count = {0};
+                log.replay((seq, addr, len) -> {
+                    count[0]++;
+                    return true;
+                });
+                assertEquals("no frames should remain after force-rotate-trim",
+                        0, count[0]);
+            }
+            // Reopen with a fresh SegmentLog; replay must visit zero frames.
+            try (SegmentLog log2 = SegmentLog.open(tmpDir, 1L << 20)) {
+                int[] count = {0};
+                log2.replay((seq, addr, len) -> {
+                    count[0]++;
+                    return true;
+                });
+                assertEquals(
+                        "acked-and-trimmed frames must not replay on restart",
+                        0, count[0]);
+                assertEquals("nextSeq must round-trip", 5L, log2.nextSeq());
             }
         });
     }