fix(qwp): skip-and-self-ack the recovered orphaned deferred tail

Closes the resurrect-partial-transaction hole documented in the
deferred-ack series: a producer that crashes (or closes)
mid-transaction leaves FLAG_DEFER_COMMIT frames with no covering
commit frame at the top of its disk-recovered SF log. Replaying them
would let a commit-bearing frame from the NEW session commit them --
half a dead transaction resurrected, violating transaction=on
atomicity. Previously detected + WARN'd only.

Mechanism (CursorWebSocketSendLoop):
- The loop adopts the engine's recovered orphan range
  [recoveredCommitBoundaryFsn+1 .. recoveredOrphanTipFsn] at
  construction; the BackgroundDrainer path builds through the same
  constructor and gets identical containment.
- trySendOne stops the cursor at the tail: orphan frames are never
  transmitted.
- tryRetireOrphanTail retires the tail with a cumulative
  self-acknowledge once every frame below it is server-acked (legal:
  the tail was never sent, no wire sequence references it). Crash-safe
  and idempotent -- dying before the self-ack re-detects the same
  range at next recovery.
- Fast path (common crash profile: acks current, only the in-flight
  transaction unacked; trivially when the whole log is the tail): the
  tail retires in positionCursorForStart before any connection exists.
  Zero wire cost, no reconnect.
- Slow path (unacked committed frames below the tail): they replay
  first; on retirement the connection recycles exactly once so the
  linear wireSeq<->FSN mapping re-anchors past the FSN gap. swapClient
  also re-attempts retirement so a reconnect never replays into the
  tail.

Positive-identification hardening (found by the drainer test suite):
the recovery walk classifies a frame as deferred ONLY when it parses
as a QWP message -- payload >= HEADER_SIZE and u32 'QWP1' magic at
offset 0 -- before reading the flags byte. Anything unidentifiable is
a retirement barrier, never trimmable: misclassifying an unknown
frame as deferred would silently discard data that should replay (the
drainer tests' filler payloads had bit 0 set at offset 5 and were
being trimmed wholesale).

Tests: CursorWebSocketSendLoopOrphanTailTest covers recovery
detection (orphan tail present/absent), the zero-send fast path, and
the slow path end-to-end (replay below tail, single recycle, correct
ack attribution across the FSN gap for new-session frames). Full
client suite: 2509 green.

Author: bluestreak01

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

@@ -96,6 +96,14 @@ public final class CursorSendEngine implements QuietCloseable {
     // covers them. Read by the sender's close-time drain to avoid waiting on
     // acks that cannot arrive.
     private long recoveredCommitBoundaryFsn = -1L;
+    // FSN of the last frame of a recovered orphaned deferred tail, or -1 when
+    // the recovered ring has no such tail. When >= 0, frames
+    // [recoveredCommitBoundaryFsn + 1 .. recoveredOrphanTipFsn] all carry
+    // FLAG_DEFER_COMMIT with no covering commit frame -- an aborted
+    // transaction. The send loop must never transmit them; it retires the
+    // range with a cumulative self-acknowledge once everything below is
+    // server-acked (CursorWebSocketSendLoop.tryRetireOrphanTail).
+    private long recoveredOrphanTipFsn = -1L;
     // Engine-owned mmap'd watermark file. {@code null} in memory mode and
     // in disk mode if open() failed (we proceed without it; recovery just
     // falls back to lowestBase - 1). Lifetime tied to the engine: opened
@@ -279,15 +287,16 @@ private CursorSendEngine(String sfDir, long segmentSizeBytes, SegmentManager man
                 // below. Computed before the I/O loop or producer append.
                 this.recoveredCommitBoundaryFsn = recovered.findLastFsnWithoutPayloadFlag(
                         io.questdb.client.cutlass.qwp.protocol.QwpConstants.HEADER_OFFSET_FLAGS,
-                        io.questdb.client.cutlass.qwp.protocol.QwpConstants.FLAG_DEFER_COMMIT
+                        io.questdb.client.cutlass.qwp.protocol.QwpConstants.FLAG_DEFER_COMMIT,
+                        io.questdb.client.cutlass.qwp.protocol.QwpConstants.MAGIC_MESSAGE,
+                        io.questdb.client.cutlass.qwp.protocol.QwpConstants.HEADER_SIZE
                 );
                 if (publishedFsn >= 0 && recoveredCommitBoundaryFsn < publishedFsn) {
+                    this.recoveredOrphanTipFsn = publishedFsn;
                     LOG.warn("recovered SF log ends with {} deferred frame(s) whose transaction was never "
-                                    + "committed [commitBoundaryFsn={}, publishedFsn={}]. On replay these frames "
-                                    + "re-append rows the server will only commit when THIS session sends a "
-                                    + "commit-bearing frame -- which would resurrect a partial transaction. "
-                                    + "If the prior transaction must stay aborted, clear the sf_dir before "
-                                    + "restarting the producer.",
+                                    + "committed [commitBoundaryFsn={}, publishedFsn={}]. The tail belongs to an "
+                                    + "aborted transaction: it will never be transmitted and its slots are "
+                                    + "retired (trimmed) once every frame below it is server-acked.",
                             publishedFsn - Math.max(recoveredCommitBoundaryFsn, -1L),
                             recoveredCommitBoundaryFsn, publishedFsn);
                 }
@@ -641,6 +650,15 @@ public long recoveredCommitBoundaryFsn() {
         return recoveredCommitBoundaryFsn;
     }
 
+    /**
+     * FSN of the last frame of a recovered orphaned deferred tail, or
+     * {@code -1} when none. See {@link #recoveredCommitBoundaryFsn()}: the
+     * orphan range is {@code [recoveredCommitBoundaryFsn() + 1 .. this]}.
+     */
+    public long recoveredOrphanTipFsn() {
+        return recoveredOrphanTipFsn;
+    }
+
     /**
      * Unlinks every {@code .sfa} file under {@code dir}. Called only on
      * clean shutdown when the ring confirms every published FSN has been

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

@@ -275,6 +275,18 @@ public final class CursorWebSocketSendLoop implements QuietCloseable {
     // to -1 once trySendOne has caught up past it. Used to count replay
     // frames without a per-frame branch on the steady-state path.
     private long replayTargetFsn = -1L;
+    // Recovered orphaned deferred tail: frames [orphanSkipStartFsn ..
+    // orphanSkipTipFsn] carry FLAG_DEFER_COMMIT with no covering commit frame
+    // -- a transaction whose commit was never published (producer crashed or
+    // closed mid-transaction). They must never reach the wire: a
+    // commit-bearing frame from THIS session would commit them server-side
+    // and resurrect a partial transaction. trySendOne stops the cursor at the
+    // tail; tryRetireOrphanTail retires it with a cumulative self-acknowledge
+    // once everything below is server-acked. -1 when none/retired. Written by
+    // the constructor and start() (user thread, before the I/O thread exists)
+    // and by the I/O thread afterwards -- never concurrently.
+    private long orphanSkipStartFsn = -1L;
+    private long orphanSkipTipFsn = -1L;
     // Poison-frame detector state (I/O thread only). poisonFsn is the FSN of the
     // frame implicated by the most recent server-active rejection: the NACK-named
     // frame, or the OK-level head-of-line frame (highestOkFsn+1) for a
@@ -478,6 +490,15 @@ public CursorWebSocketSendLoop(WebSocketClient client, CursorSendEngine engine,
         // hands null and lets the I/O thread connect — hasEverConnected
         // stays false until swapClient sees its first success.
         this.hasEverConnected = client != null;
+        // Adopt the engine's recovered orphaned-deferred-tail range (if any).
+        // See the field docs on orphanSkipStartFsn/orphanSkipTipFsn; the
+        // BackgroundDrainer path builds its loop through this same
+        // constructor, so drained orphan slots get the identical containment.
+        long orphanTip = engine.recoveredOrphanTipFsn();
+        if (orphanTip >= 0) {
+            this.orphanSkipStartFsn = engine.recoveredCommitBoundaryFsn() + 1L;
+            this.orphanSkipTipFsn = orphanTip;
+        }
     }
 
     /**
@@ -1759,6 +1780,11 @@ private void swapClient(WebSocketClient newClient) {
                 // best-effort
             }
         }
+        // A recovered orphaned deferred tail may be retirable now (e.g. the
+        // acks covering everything below it arrived just before the drop).
+        // Retiring before computing replayStart anchors the new connection
+        // past the tail instead of replaying into it.
+        tryRetireOrphanTail();
         long replayStart = engine.ackedFsn() + 1L;
         this.fsnAtZero = replayStart;
         this.nextWireSeq = 0L;
@@ -1794,6 +1820,34 @@ private boolean tryReceiveAcks() {
      * scheduling fairness.
      */
     private boolean trySendOne() {
+        if (orphanSkipTipFsn >= 0 && fsnAtZero + nextWireSeq >= orphanSkipStartFsn) {
+            // The send cursor reached the orphaned deferred tail. Its frames
+            // belong to an aborted transaction and must never be transmitted
+            // (a commit-bearing frame from this session would commit them --
+            // partial-transaction resurrection).
+            if (!tryRetireOrphanTail()) {
+                // Frames below the tail still await server acks; keep ticking
+                // (tryReceiveAcks runs every loop iteration) until the ack
+                // watermark reaches the tail's lower edge.
+                return false;
+            }
+            if (nextWireSeq == 0) {
+                // Nothing sent on this connection yet: re-anchor in place past
+                // the retired tail. The wireSeq<->FSN mapping is untouched
+                // because no wire sequence has been consumed.
+                positionCursorForStart();
+                return true;
+            }
+            // Frames were already sent on this connection: the linear
+            // fsnAtZero + wireSeq mapping cannot express the gap the retired
+            // tail leaves behind. Recycle; swapClient re-anchors the mapping
+            // at ackedFsn + 1 = the first frame past the tail. One reconnect,
+            // once per recovery -- and only on the slow path (unacked
+            // committed frames existed below the tail).
+            fail(new LineSenderException(
+                    "recycling connection after retiring orphaned deferred tail (wire-seq realignment)"));
+            return false;
+        }
         long pub = sendingSegment.publishedOffset();
         if (sendOffset >= pub) {
             // Nothing more in the current segment. If it's a sealed segment
@@ -1859,12 +1913,53 @@ private boolean trySendOne() {
      * real I/O thread + WebSocket.
      */
     void positionCursorForStart() {
+        // Fast path for a recovered orphaned deferred tail: when everything
+        // below the tail is already acked (the common crash profile -- acks
+        // were flowing until the crash, only the in-flight transaction is
+        // unacked; and trivially when the WHOLE recovered log is the tail),
+        // the tail retires before any connection exists and the cursor
+        // starts past it. Zero wire cost, no recycle.
+        tryRetireOrphanTail();
         long replayStart = engine.ackedFsn() + 1L;
         this.fsnAtZero = replayStart;
         this.nextWireSeq = 0L;
         positionCursorAt(replayStart);
     }
 
+    /**
+     * Retires the recovered orphaned deferred tail once retirable.
+     * <p>
+     * The tail's frames all carry FLAG_DEFER_COMMIT with no covering commit
+     * frame: the producer died (or closed) mid-transaction, so the
+     * transaction is aborted by definition. Retirement is a cumulative
+     * self-acknowledge of the tail's top FSN -- legal once every frame BELOW
+     * the tail is server-acked, because the tail frames were never
+     * transmitted and no wire sequence references them. The freed slots are
+     * trimmed by the normal ack-driven machinery.
+     * <p>
+     * Crash-safe: dying before the self-ack leaves the tail in place; the
+     * next recovery re-detects the same range and retries. Idempotent once
+     * retired.
+     *
+     * @return true when no orphan tail remains (retired now, previously, or
+     *         never existed); false while frames below the tail are unacked
+     */
+    private boolean tryRetireOrphanTail() {
+        if (orphanSkipTipFsn < 0) {
+            return true;
+        }
+        if (engine.ackedFsn() < orphanSkipStartFsn - 1L) {
+            return false;
+        }
+        LOG.warn("retiring orphaned deferred tail: {} frame(s) [fsn {}..{}] belong to a transaction "
+                        + "whose commit was never published; aborting them (never transmitted, slots trimmed)",
+                orphanSkipTipFsn - orphanSkipStartFsn + 1, orphanSkipStartFsn, orphanSkipTipFsn);
+        engine.acknowledge(orphanSkipTipFsn);
+        orphanSkipStartFsn = -1L;
+        orphanSkipTipFsn = -1L;
+        return true;
+    }
+
     /**
      * Factory used by the I/O loop to build a fresh, connected, upgraded
      * {@link WebSocketClient} after a wire failure. Implementations close

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

@@ -446,23 +446,36 @@ public long tryAppend(long payloadAddr, int payloadLen) {
     /**
      * Walks every published frame in this segment and returns the FSN of the
      * LAST frame whose payload does NOT carry the given flag bit, or {@code -1}
-     * when every frame carries it (or the segment is empty). Frames shorter
-     * than {@code flagsOffset + 1} bytes are treated as not carrying the flag.
+     * when every frame carries it (or the segment is empty).
+     * <p>
+     * A frame counts as carrying the flag ONLY when it positively parses as a
+     * message of the expected protocol: payload at least {@code minPayloadLen}
+     * bytes AND the little-endian u32 at payload offset 0 equals
+     * {@code headerMagic} AND the byte at {@code flagsOffset} has
+     * {@code flagMask} set. Anything else -- short frames, foreign payloads,
+     * magic mismatches -- counts as NOT carrying the flag. This direction is
+     * deliberate: the caller retires (trims) frames ABOVE the returned FSN,
+     * so a frame we cannot positively identify must act as a retirement
+     * barrier, never as trimmable. Misclassifying an unknown frame as
+     * deferred would silently discard data that should replay.
      * <p>
      * Producer-thread only, and only meaningful before new appends race the
      * walk (recovery time). Used to locate the last commit-bearing QWP frame
      * below a potentially orphaned FLAG_DEFER_COMMIT tail: frames above the
      * returned FSN all carry the flag, i.e. they belong to a transaction whose
      * commit frame was never published.
      */
-    public long findLastFrameFsnWithoutPayloadFlag(int flagsOffset, int flagMask) {
+    public long findLastFrameFsnWithoutPayloadFlag(int flagsOffset, int flagMask, int headerMagic, int minPayloadLen) {
         long best = -1L;
         long off = HEADER_SIZE;
         long frames = frameCount;
         for (long i = 0; i < frames; i++) {
             int payloadLen = Unsafe.getUnsafe().getInt(mmapAddress + off + 4);
-            boolean flagSet = payloadLen > flagsOffset
-                    && (Unsafe.getUnsafe().getByte(mmapAddress + off + FRAME_HEADER_SIZE + flagsOffset) & flagMask) != 0;
+            long payload = mmapAddress + off + FRAME_HEADER_SIZE;
+            boolean flagSet = payloadLen >= minPayloadLen
+                    && payloadLen > flagsOffset
+                    && Unsafe.getUnsafe().getInt(payload) == headerMagic
+                    && (Unsafe.getUnsafe().getByte(payload + flagsOffset) & flagMask) != 0;
             if (!flagSet) {
                 best = baseSeq + i;
             }

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

@@ -521,17 +521,21 @@ public synchronized MmapSegment firstSealed() {
      * potentially orphaned FLAG_DEFER_COMMIT tail left behind by a producer
      * that crashed (or closed) mid-transaction. Call before the I/O loop and
      * producer start appending; the walk is not synchronized against appends
-     * into the active segment.
+     * into the active segment. See
+     * {@link MmapSegment#findLastFrameFsnWithoutPayloadFlag} for the
+     * positive-identification contract: frames that do not parse as protocol
+     * messages count as commit-bearing (retirement barriers), never as
+     * trimmable.
      */
-    public synchronized long findLastFsnWithoutPayloadFlag(int flagsOffset, int flagMask) {
+    public synchronized long findLastFsnWithoutPayloadFlag(int flagsOffset, int flagMask, int headerMagic, int minPayloadLen) {
         long best = -1L;
         for (int i = 0, n = sealedSegments.size(); i < n; i++) {
-            long fsn = sealedSegments.get(i).findLastFrameFsnWithoutPayloadFlag(flagsOffset, flagMask);
+            long fsn = sealedSegments.get(i).findLastFrameFsnWithoutPayloadFlag(flagsOffset, flagMask, headerMagic, minPayloadLen);
             if (fsn > best) {
                 best = fsn;
             }
         }
-        long fsn = active.findLastFrameFsnWithoutPayloadFlag(flagsOffset, flagMask);
+        long fsn = active.findLastFrameFsnWithoutPayloadFlag(flagsOffset, flagMask, headerMagic, minPayloadLen);
         return Math.max(best, fsn);
     }