fix(qwp): stop close() waiting on acks of uncommitted deferred frames

The server (as of the companion core fix) withholds cumulative OK acks
for FLAG_DEFER_COMMIT frames until their group-closing commit lands:
acking them earlier let the store-and-forward client trim slots whose
rows the server rolls back on any error, demote, or disconnect --
silent data loss (#7144's stated replay contract, previously
unenforced).

Client-side consequences fixed here:

- QwpWebSocketSender tracks lastCommitBoundaryFsn: the FSN of the last
  commit-bearing (non-deferred) frame published. Updated by
  flushPendingRows, flushPendingRowsSplit, sendCommitMessage, and the
  close()-time residual seal.
- drainOnClose() now drains to min(publishedFsn, commit boundary)
  instead of publishedFsn. Waiting for acks of an uncommitted deferred
  tail could only ever time out (300s hangs in
  testDeferredCommitConnectionDropRollsBack). Abandoning frames is
  WARN-logged with an actionable message (flush() to commit, or the
  abort is intentional).
- CursorSendEngine computes recoveredCommitBoundaryFsn at disk
  recovery: the last commit-bearing frame below a potentially orphaned
  deferred tail left by a producer that died mid-transaction. Close
  drains combine it with the session boundary. Recovery WARNs when an
  orphaned deferred tail exists, because replaying it into a NEW
  session's commit would resurrect a partial transaction -- the
  skip-and-self-ack replay policy for that tail is a documented
  follow-up.
- MmapSegment/SegmentRing grow a read-only recovery-time frame walk
  (findLastFsnWithoutPayloadFlag) that locates the boundary by peeking
  the QWP flags byte of each published frame.

Full client suite passes (2505 tests).

Author: bluestreak01

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

@@ -253,6 +253,13 @@ public class QwpWebSocketSender implements Sender {
     private int errorInboxCapacity = SenderErrorDispatcher.DEFAULT_CAPACITY;
     private long firstPendingRowTimeNanos;
     private boolean hasDeferredMessages;
+    // FSN of the last commit-bearing (non-FLAG_DEFER_COMMIT) frame this session
+    // published, or -1 when none. Frames above it are deferred and uncommitted:
+    // the server withholds their acks by design (their rows are rolled back on
+    // any error, demote, or disconnect), so close-time drains must never wait
+    // for them. Updated on every non-deferred publish; combined with the
+    // engine's recovered boundary in drainOnClose.
+    private long lastCommitBoundaryFsn = -1L;
     // Stickys true once any successful FOREGROUND connect has happened
     // (background drainer connects never set it). Drives the
     // CONNECTED-vs-RECONNECTED-vs-FAILED_OVER classification at the success
@@ -1076,6 +1083,9 @@ public void close() {
                     }
                     if (activeBuffer != null && activeBuffer.hasData()) {
                         sealAndSwapBuffer();
+                        if (!deferCommit) {
+                            lastCommitBoundaryFsn = cursorEngine.publishedFsn();
+                        }
                     }
                     // 2) Safety-net rethrow: surface the latched terminal
                     //    error only when no other channel has already
@@ -3149,7 +3159,22 @@ private void drainOnClose(boolean errorOwnedByCustomHandler) {
         if (closeFlushTimeoutMillis <= 0L) {
             return;
         }
-        long target = cursorEngine.publishedFsn();
+        long published = cursorEngine.publishedFsn();
+        // Never wait for uncommitted deferred frames: the server withholds
+        // their acks by design (FLAG_DEFER_COMMIT rows are rolled back on
+        // error/demote/disconnect and must stay replayable client-side), so a
+        // drain targeting them can only time out. The drain target is the last
+        // commit-bearing frame this session published, or -- for a ring
+        // recovered from disk with no commit published this session -- the
+        // recovered commit boundary.
+        long boundary = Math.max(lastCommitBoundaryFsn, cursorEngine.recoveredCommitBoundaryFsn());
+        long target = Math.min(published, boundary);
+        if (target < published) {
+            LOG.warn("close() abandoning {} uncommitted deferred frame(s) [commitBoundaryFsn={}, publishedFsn={}] "
+                            + "-- their transaction was never committed; the server rolls their rows back. "
+                            + "Call flush() before close() to commit, or ignore if the abort is intentional.",
+                    published - target, target, published);
+        }
         if (cursorEngine.ackedFsn() >= target) {
             return;
         }
@@ -3169,7 +3194,7 @@ private void drainOnClose(boolean errorOwnedByCustomHandler) {
                 LOG.warn("close() drain timed out after {}ms [target={} acked={}], pending data may be lost",
                         closeFlushTimeoutMillis, target, acked);
                 throw new LineSenderException("close() drain timed out after ")
-                        .put(closeFlushTimeoutMillis).put(" ms [publishedFsn=")
+                        .put(closeFlushTimeoutMillis).put(" ms [targetFsn=")
                         .put(target).put(", ackedFsn=").put(acked)
                         .put("] - server did not acknowledge ")
                         .put(target - acked)
@@ -3430,6 +3455,9 @@ private void flushPendingRows(boolean deferCommit) {
         sealAndSwapBuffer();
 
         hasDeferredMessages = deferCommit;
+        if (!deferCommit) {
+            lastCommitBoundaryFsn = cursorEngine.publishedFsn();
+        }
 
         resetTableBuffersAfterFlush(keys);
     }
@@ -3502,6 +3530,11 @@ private void flushPendingRowsSplit(ObjList<CharSequence> keys, boolean deferComm
 
         encoder.setDeferCommit(false);
         hasDeferredMessages = deferCommit;
+        if (!deferCommit) {
+            // The last message of the split carried no defer flag -- it
+            // committed the whole group.
+            lastCommitBoundaryFsn = cursorEngine.publishedFsn();
+        }
         resetTableBuffersAfterFlush(keys);
     }
 
@@ -3543,6 +3576,7 @@ private void sendCommitMessage() {
         activeBuffer.incrementRowCount();
         sealAndSwapBuffer();
         hasDeferredMessages = false;
+        lastCommitBoundaryFsn = cursorEngine.publishedFsn();
     }
 
     private void resetSymbolDictStateForNewConnection() {

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

@@ -88,6 +88,14 @@ public final class CursorSendEngine implements QuietCloseable {
     // full symbol-dict delta), so producer-side schema reset on recovery
     // is not required.
     private final boolean wasRecoveredFromDisk;
+    // FSN of the last commit-bearing (non-FLAG_DEFER_COMMIT) frame found in a
+    // ring recovered from disk, or -1 for fresh/memory rings and recovered
+    // rings whose every frame is deferred. Frames above this FSN in the
+    // recovered ring belong to a transaction whose commit frame was never
+    // published; the server will never ack them until some later commit
+    // covers them. Read by the sender's close-time drain to avoid waiting on
+    // acks that cannot arrive.
+    private long recoveredCommitBoundaryFsn = -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
@@ -260,6 +268,29 @@ private CursorSendEngine(String sfDir, long segmentSizeBytes, SegmentManager man
                 if (seed >= 0) {
                     recovered.acknowledge(seed);
                 }
+                // Locate the last commit-bearing frame below a potentially
+                // orphaned FLAG_DEFER_COMMIT tail. A producer that crashed (or
+                // closed) mid-transaction leaves deferred frames with no
+                // covering commit frame at the top of the ring. The server
+                // never acks uncommitted deferred frames, so (a) close-time
+                // drains must not wait for them (see the sender's
+                // drainOnClose), and (b) replaying them into a NEW session's
+                // commit would resurrect half a transaction -- see the WARN
+                // 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
+                );
+                if (publishedFsn >= 0 && recoveredCommitBoundaryFsn < 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.",
+                            publishedFsn - Math.max(recoveredCommitBoundaryFsn, -1L),
+                            recoveredCommitBoundaryFsn, publishedFsn);
+                }
             } else {
                 // Fresh start with no recovered segments. Any stale
                 // watermark from a prior fully-drained session refers
@@ -600,6 +631,16 @@ public boolean wasRecoveredFromDisk() {
         return wasRecoveredFromDisk;
     }
 
+    /**
+     * FSN of the last commit-bearing frame in a disk-recovered ring, or
+     * {@code -1} for fresh/memory rings. Frames above it are an orphaned
+     * deferred tail (transaction never committed) that the server will not
+     * ack until a later commit-bearing frame covers them.
+     */
+    public long recoveredCommitBoundaryFsn() {
+        return recoveredCommitBoundaryFsn;
+    }
+
     /**
      * 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/MmapSegment.java

@@ -443,6 +443,34 @@ public long tryAppend(long payloadAddr, int payloadLen) {
         return offset;
     }
 
+    /**
+     * 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.
+     * <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) {
+        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;
+            if (!flagSet) {
+                best = baseSeq + i;
+            }
+            off += FRAME_HEADER_SIZE + payloadLen;
+        }
+        return best;
+    }
+
     /**
      * Number of frames written since {@link #create} (or recovered by
      * {@link #openExisting}). Used by {@code SegmentRing} to compute

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

@@ -510,6 +510,31 @@ public synchronized MmapSegment firstSealed() {
     }
 
     /** Active segment -- exposed for the I/O thread's "send next batch" path. */
+    /**
+     * Walks every published frame in the ring (sealed segments plus the active
+     * segment) and returns the FSN of the LAST frame whose payload does NOT
+     * carry the given flag bit, or {@code -1} when every published frame
+     * carries it (or the ring is empty). All frames above the returned FSN
+     * carry the flag.
+     * <p>
+     * Recovery-time helper: locates the last commit-bearing QWP frame below a
+     * 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.
+     */
+    public synchronized long findLastFsnWithoutPayloadFlag(int flagsOffset, int flagMask) {
+        long best = -1L;
+        for (int i = 0, n = sealedSegments.size(); i < n; i++) {
+            long fsn = sealedSegments.get(i).findLastFrameFsnWithoutPayloadFlag(flagsOffset, flagMask);
+            if (fsn > best) {
+                best = fsn;
+            }
+        }
+        long fsn = active.findLastFrameFsnWithoutPayloadFlag(flagsOffset, flagMask);
+        return Math.max(best, fsn);
+    }
+
     public MmapSegment getActive() {
         return active;
     }