fix(client): coordinate segment trim with I/O cursor

Author: bluestreak01

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

@@ -528,6 +528,10 @@ public MmapSegment activeSegment() {
         return ring.getActive();
     }
 
+    MmapSegment pinActiveSegment() {
+        return ring.pinActiveSegment();
+    }
+
     /**
      * Append the payload, blocking up to {@link #appendDeadlineNanos} when
      * the cursor ring is at its memory/disk cap and waiting for ACK-driven
@@ -613,7 +617,7 @@ public synchronized void close() {
         closed = true;
         // Capture drain state BEFORE closing the ring — once the ring is
         // closed, its accessors aren't safe to read. The active segment is
-        // never trimmed by drainTrimmable (only sealed segments are), so
+        // never trim-eligible (only sealed segments are), so
         // when everything published has been acked we have to unlink the
         // residual .sfa files here. Without this, the next sender (or a
         // drainer adopting this slot) would replay already-acked data
@@ -1191,6 +1195,10 @@ public MmapSegment findSegmentContaining(long fsn) {
         return ring.findSegmentContaining(fsn);
     }
 
+    MmapSegment pinSegmentContaining(long fsn) {
+        return ring.pinSegmentContaining(fsn);
+    }
+
     /**
      * Pass-through to {@link SegmentRing#firstSealed()}.
      */
@@ -1215,6 +1223,14 @@ public MmapSegment nextSealedAfter(MmapSegment current) {
         return ring.nextSealedAfter(current);
     }
 
+    MmapSegment advancePinnedSegment(MmapSegment current) {
+        return ring.advancePinnedSegment(current);
+    }
+
+    void releasePinnedSegment(MmapSegment current) {
+        ring.releasePinnedSegment(current);
+    }
+
     /**
      * I/O thread accessor: highest FSN whose frame is fully written.
      */

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

@@ -896,6 +896,9 @@ public synchronized void close() {
             }
             ioThread = null;
         }
+        // Covers close-before-start and start failures; normal I/O-thread
+        // exit already released the same pin in its finally block.
+        releaseSendingSegment();
         // Close the current client. After a reconnect, swapClient has
         // replaced the original (and closed it); the owner only retains
         // the stale pre-reconnect reference. Without closing the live
@@ -1102,17 +1105,24 @@ public synchronized void start() {
         // 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();
+        try {
+            positionCursorForStart();
+        } catch (Throwable t) {
+            running = false;
+            releaseSendingSegment();
+            shutdownLatch.countDown();
+            throw t;
+        }
         Thread t = new Thread(this::ioLoop, "qdb-cursor-ws-io");
         t.setDaemon(true);
         try {
             t.start();
         } catch (Throwable th) {
             // Thread.start() failed (e.g. native stack alloc OOM). ioLoop
-            // never ran, so its finally{shutdownLatch.countDown()} never
-            // fires. Release the latch and reset state so a subsequent
-            // close() doesn't block on a thread that doesn't exist.
+            // never ran, so its finally block cannot release the segment pin
+            // or count down the latch.
             running = false;
+            releaseSendingSegment();
             shutdownLatch.countDown();
             throw th;
         }
@@ -1132,35 +1142,17 @@ private PendingDurableEntry acquirePendingEntry() {
      * else the active). Returns the same segment if it's still being written
      * (we're on the active and just need to wait for more publishedFsn).
      * <p>
-     * Uses {@link CursorSendEngine#nextSealedAfter} so we never have to
-     * snapshot the full sealed list — important when the producer outpaces
-     * the I/O thread and the sealed list can grow to thousands of entries
-     * (cursor SF lets the producer fan out at memory speed; the wire path
-     * catches up at WebSocket speed).
+     * The ring switches the single I/O pin atomically with choosing the next
+     * segment, so trim can never unmap either side of the handoff. No sealed
+     * list snapshot is needed when the producer outpaces the wire path.
      */
     private MmapSegment advanceSegment() {
         MmapSegment current = sendingSegment;
-        MmapSegment liveActive = engine.activeSegment();
-        if (current == liveActive) {
-            // We're on the active — there's no "next", just wait for more
-            // bytes to be published into it. Caller's sendOne will see
-            // publishedOffset > sendOffset eventually and resume.
-            return current;
+        MmapSegment next = engine.advancePinnedSegment(current);
+        if (next != current) {
+            sendOffset = MmapSegment.HEADER_SIZE;
         }
-        sendOffset = MmapSegment.HEADER_SIZE;
-        MmapSegment next = engine.nextSealedAfter(current);
-        if (next != null) {
-            return next;
-        }
-        // current was the newest sealed (no later sealed exists). If it's
-        // still in the sealed list, the next segment must be the active;
-        // if it's been trimmed out from under us, fall back to the oldest
-        // remaining sealed before resorting to the active.
-        next = engine.firstSealed();
-        if (next != null && next.baseSeq() > current.baseSeq()) {
-            return next;
-        }
-        return liveActive;
+        return next;
     }
 
     private void applyDurableAck() {
@@ -1760,6 +1752,9 @@ private void ioLoop() {
                 }
             }
         } finally {
+            // Release native-segment lifetime before publishing I/O-thread
+            // completion or running delegated engine cleanup.
+            releaseSendingSegment();
             // Last act of the I/O thread: dispose of whatever client it
             // holds. This is the airtight half of the close()-vs-reconnect
             // race — when close()'s latch await is interrupted (drainer pool
@@ -1809,7 +1804,7 @@ private void ioLoop() {
      * loop will then wait until the producer publishes more bytes.
      */
     private void positionCursorAt(long targetFsn) {
-        MmapSegment seg = engine.findSegmentContaining(targetFsn);
+        MmapSegment seg = engine.pinSegmentContaining(targetFsn);
         if (seg == null) {
             // No segment currently advertises targetFsn. That normally means
             // targetFsn is just past publishedFsn and there is nothing to
@@ -1818,14 +1813,14 @@ private void positionCursorAt(long targetFsn) {
             // The producer is concurrent with this I/O thread, though. It can
             // publish targetFsn after the first findSegmentContaining() returns
             // null but before or during the active-tip snapshot below.
-            sendingSegment = engine.activeSegment();
+            sendingSegment = engine.pinActiveSegment();
             sendOffset = sendingSegment.publishedOffset();
             // The publishedOffset read is the producer's volatile publish
             // barrier. If it saw the new frame bytes, the frameCount write that
             // makes targetFsn discoverable is also visible, so a second lookup
             // must now find it. If the producer publishes later, sendOffset is
             // still at the old tip and trySendOne() will send the frame normally.
-            seg = engine.findSegmentContaining(targetFsn);
+            seg = engine.pinSegmentContaining(targetFsn);
             if (seg != null) {
                 positionCursorInSegment(seg, targetFsn);
             }
@@ -1905,6 +1900,14 @@ private void releasePendingEntry(PendingDurableEntry e) {
         pendingDurablePool.addFirst(e);
     }
 
+    private void releaseSendingSegment() {
+        MmapSegment segment = sendingSegment;
+        if (segment != null) {
+            engine.releasePinnedSegment(segment);
+            sendingSegment = null;
+        }
+    }
+
     /**
      * Send a WebSocket PING to prod the server into flushing pending
      * STATUS_DURABLE_ACK frames, but only when the throttle interval has