fix(ilp): five SF correctness fixes from PR-17 review

Each fix is paired with a red-then-green regression test.

C1 — SegmentLog.rotate() partial failure deadlock
  When rotate's allocNativePath OOMs or createActive fails after the
  rename succeeds, `active` is left pointing at a sealed segment with
  fd=-1. A subsequent small append that fits under the cap bypasses
  the rotate trigger and falls through to ff.write(fd=-1, ...) which
  returns -1 and is wrapped as the recoverable SfDiskFullException.
  The I/O thread retries forever (disk-full backpressure path) and
  the user thread blocks in flush() — silent deadlock. Guard added at
  the top of append() that throws a fatal SfException when active is
  in the post-rotate sealed/fd=-1 state.

C2 — Symbol-delta watermark not reset on reconnect
  resetSchemaStateForNewConnection cleared maxSentSchemaId and
  per-table schema ids but left maxSentSymbolId and
  currentBatchMaxSymbolId untouched. The encoder's first
  post-reconnect batch then shipped a delta dictionary that excluded
  every symbol id <= the old server's high-water mark; column refs
  into the new server's empty dictionary decoded as garbage (or were
  rejected). Both watermarks are now reset alongside the schema
  state.

C3 — trim() forgets sealed segments when remove() fails
  trimSealedSegments discarded the boolean return of ff.remove(). On
  Windows sharing-violation under antivirus, transient NFS errors,
  ESTALE, etc., the file stayed on disk while bytesOnDiskCache was
  decremented and the segment was dropped from the in-memory list.
  Failure modes: (a) bytesOnDisk underreports reality, so
  sf_max_total_bytes stops being an enforceable cap; (b) on next
  process start scanDirectory rediscovers the orphan .sfs and
  re-ships its already-acked frames to the new server. Failed
  removes now keep the segment in the list with a removePending
  flag — bytesOnDiskCache stays honest, replay() skips removePending
  segments (so already-acked frames don't re-ship), the next trim()
  retries naturally, and close() does a last-chance retry too.

C4 — Future server ACK can delete unsent SF data
  InFlightWindow.acknowledgeUpTo clamps incoming server sequence at
  highestSent; ResponseHandler.onBinaryMessage was passing the raw
  uncapped sequence into segmentLog.trim(fsnAtZero + sequence) with
  no symmetric clamp. A buggy/replayed/malformed server ACK with a
  sequence beyond what the client had sent drove SegmentLog.trim
  past every real lastSeq, force-rotating the active segment and
  unlinking every sealed segment whose lastSeq <= the bogus value —
  including frames mid-replay the new server had never seen.
  Permanent silent data loss. The trim path now clamps the sequence
  at nextBatchSequence-1 (mirroring the InFlightWindow cap) and
  emits a WARN when the cap fires.

C5 — Replay window-wait spin hangs after mid-replay socket drop
  replayPersistedFrames's window-wait spin only called
  tryReceiveAcks while client.isConnected() returned true. When the
  server dropped mid-replay, isConnected went false, no ACKs could
  arrive, hasWindowSpace stayed false, and the spin ran forever —
  preventing the outer state machine from running another
  doReconnectCycle and blocking flush()/close() until the user
  signalled shutdown. Compounding: replayPersistedFrames swallowed
  internal failures via failConnection(non-fatal) and returned
  normally, so doReconnectCycle returned true and ioLoop cleared
  reconnectRequested — losing the failure. The spin now exits on
  !isConnected or reconnectRequested; doReconnectCycle clears the
  stale reconnectRequested before replay (so the freshly-reconnected
  spin doesn't bail) and re-checks it after replay (so internal
  failures propagate to the outer loop's backoff/retry).

Tests:
  SegmentLogTest:
    + testRotateOomThenSmallAppendThrowsFatalNotDiskFull (C1)
    + testTrimRemoveFailureMustNotForgetSealedSegment   (C3)
  SfIntegrationTest:
    + testReconnectResetsSymbolWatermark               (C2)
    + testFutureAckMustNotTrimUnsentSfData             (C4)
    + testReplayMustNotHangWhenConnectionDropsMidReplay (C5)

All 1988 client tests pass.

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

@@ -1968,6 +1968,16 @@ private int countNonEmptyTables(ObjList<CharSequence> keys) {
     private void resetSchemaStateForNewConnection() {
         maxSentSchemaId = -1;
         nextSchemaId = 0;
+        // The new server has an empty symbol dictionary. The encoder's
+        // delta-dictionary range is computed as
+        //   deltaStart = maxSentSymbolId + 1
+        //   deltaCount = max(0, currentBatchMaxSymbolId - maxSentSymbolId)
+        // so a non-reset watermark would skip every symbol id <= the old
+        // server's confirmed max, leaving column refs into a dictionary the
+        // new server has never seen. Reset both so the next batch ships a
+        // delta starting at id 0 covering every referenced symbol.
+        maxSentSymbolId = -1;
+        currentBatchMaxSymbolId = -1;
 
         ObjList<CharSequence> keys = tableBuffers.keys();
         for (int i = 0, n = keys.size(); i < n; i++) {

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

@@ -924,12 +924,29 @@ private boolean doReconnectCycle(long sleepMs) {
         }
         long oldest = segmentLog.oldestSeq();
         fsnAtZero = oldest >= 0 ? oldest : segmentLog.nextSeq();
+        // Clear the stale reconnectRequested flag BEFORE replay, otherwise
+        // replay's window-wait spin would exit immediately on the new
+        // connection (it checks !reconnectRequested as a bail-out signal).
+        // failConnection in onClose / onError will set it again if the new
+        // connection also fails.
+        reconnectRequested = false;
         try {
             replayPersistedFrames();
         } catch (Throwable t) {
             LOG.warn("SF replay after reconnect failed: {}", t.getMessage());
             return false;
         }
+        // replayPersistedFrames swallows internal failures by calling
+        // failConnection(non-fatal), which sets reconnectRequested=true and
+        // returns normally. Without this check the caller would clear
+        // reconnectRequested and proceed as if replay succeeded — the
+        // failure would be silently lost. Surface it so the I/O loop backs
+        // off and retries.
+        if (reconnectRequested) {
+            LOG.warn("SF replay aborted mid-stream (reconnect requested) — "
+                    + "treating reconnect cycle as failed");
+            return false;
+        }
         LOG.info("SF reconnect complete");
         return true;
     }
@@ -959,13 +976,35 @@ private void replayPersistedFrames() {
                             "SF replay FSN drift: fsn=" + fsn + " expected=" + (fsnAtZero + wireSeq));
                 }
                 if (inFlightWindow != null) {
-                    while (running && !inFlightWindow.hasWindowSpace()) {
-                        if (client.isConnected()) {
-                            tryReceiveAcks();
-                        }
+                    // Wait for window space, but bail out as soon as the
+                    // connection is gone or the I/O thread has been told to
+                    // reconnect. Without these guards the loop would spin
+                    // forever once the server drops mid-replay: a closed
+                    // socket can never produce ACKs, so hasWindowSpace stays
+                    // false; the outer state machine never gets to call
+                    // doReconnectCycle because we're stuck in this lambda;
+                    // and flush()/close() block on the I/O thread that will
+                    // never make progress until the user signals shutdown.
+                    while (running
+                            && !inFlightWindow.hasWindowSpace()
+                            && !reconnectRequested
+                            && client.isConnected()) {
+                        tryReceiveAcks();
                         Thread.onSpinWait();
                     }
-                    if (!running) {
+                    if (!running || reconnectRequested || !client.isConnected()) {
+                        // Either shutdown was requested, the connection died
+                        // (so any further sends will fail anyway), or the
+                        // failure handler has already requested a reconnect.
+                        // Stop replaying and let the outer state machine
+                        // drive the next attempt. Mark reconnectRequested so
+                        // doReconnectCycle's post-replay check returns false
+                        // and the caller backs off and retries.
+                        if (running && !reconnectRequested && !client.isConnected()) {
+                            failConnection(new LineSenderException(
+                                    "Connection lost mid-replay (window full, no ACKs possible)"),
+                                    false);
+                        }
                         return false;
                     }
                     if (!inFlightWindow.tryAddInFlight(wireSeq)) {
@@ -1249,7 +1288,29 @@ public void onBinaryMessage(long payloadPtr, int payloadLen) {
                 if (segmentLog != null) {
                     // Translate wire seq → FSN. Cumulative ack of wire seq N means
                     // every FSN up to fsnAtZero+N has been applied server-side.
-                    segmentLog.trim(fsnAtZero + sequence);
+                    //
+                    // Clamp sequence at the highest wire seq the client has actually
+                    // sent on this connection (= nextBatchSequence - 1). Without this,
+                    // a misbehaving / replayed / malformed server ACK with a sequence
+                    // beyond what we sent would feed a fictitious lastSeq into
+                    // SegmentLog.trim, which would then force-rotate the active
+                    // segment and unlink every sealed segment whose lastSeq <= the
+                    // bogus value — including frames mid-replay that the new server
+                    // has never seen. Permanent silent data loss.
+                    //
+                    // This mirrors the cap that InFlightWindow.acknowledgeUpTo
+                    // already applies at line 144; the SF trim path was missing the
+                    // symmetric guard.
+                    long highestSent = nextBatchSequence - 1;
+                    if (highestSent >= 0) {
+                        long capped = Math.min(sequence, highestSent);
+                        if (capped < sequence) {
+                            LOG.warn("server ACK sequence {} exceeds highest sent "
+                                    + "wire seq {} — clamping SF trim to prevent "
+                                    + "silent data loss", sequence, highestSent);
+                        }
+                        segmentLog.trim(fsnAtZero + capped);
+                    }
                 }
                 for (int i = 0, n = response.getTableEntryCount(); i < n; i++) {
                     advanceSeqTxn(committedSeqTxns, response.getTableName(i), response.getTableSeqTxn(i));

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

@@ -31,6 +31,8 @@
 import io.questdb.client.std.ObjList;
 import io.questdb.client.std.QuietCloseable;
 import io.questdb.client.std.Unsafe;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /**
  * Segmented append-only log of opaque byte frames keyed by a monotonic 64-bit sequence number.
@@ -60,6 +62,9 @@
  */
 public final class SegmentLog implements QuietCloseable {
 
+    private static final Logger LOG = LoggerFactory.getLogger(SegmentLog.class);
+
+
     public static final long DEFAULT_MAX_BYTES_PER_SEGMENT = 64L * 1024 * 1024;
     public static final long DEFAULT_MAX_TOTAL_BYTES = Long.MAX_VALUE;
     public static final long FIRST_SEQ = 0L;
@@ -173,6 +178,21 @@ public static SegmentLog open(String dir, FilesFacade ff, long maxBytesPerSegmen
      */
     public long append(long payloadAddr, int payloadLen) {
         ensureOpen();
+        // Guard against the partial-rotate failure state (bug C1). When
+        // rotate() fails between rename and createActive (e.g. allocNativePath
+        // OOMs at the second alloc, or createActive's openCleanRW/fsync fails
+        // for the new segment), `active` is left pointing at the now-sealed
+        // segment with sealed=true and fd=-1. Without this guard, a small
+        // subsequent append that fits under the cap would bypass the rotate
+        // trigger below and fall through to ff.write(fd=-1) — which returns
+        // -1 and is wrapped as SfDiskFullException (a recoverable backpressure
+        // signal) by the short-write branch. The I/O thread would then retry
+        // forever and the user thread would deadlock in flush(). Surface a
+        // fatal SfException instead so the connection terminates cleanly.
+        if (active.sealed || active.fd < 0) {
+            throw new SfException("SegmentLog is unusable after a prior rotate failure: "
+                    + active.path + " (sealed=" + active.sealed + ", fd=" + active.fd + ")");
+        }
         if (payloadLen <= 0) {
             throw new SfException("payloadLen must be > 0, got " + payloadLen);
         }
@@ -249,7 +269,15 @@ public void fsync() {
     public void replay(FrameVisitor visitor) {
         ensureOpen();
         for (int i = 0, n = segments.size(); i < n; i++) {
-            if (!replaySegment(segments.getQuick(i), visitor)) {
+            Segment s = segments.getQuick(i);
+            // Skip segments whose disk file we tried (and failed) to remove
+            // on a previous trim. Their frames were acked by the server —
+            // re-shipping them on the new connection would produce silent
+            // duplicate writes.
+            if (s.removePending) {
+                continue;
+            }
+            if (!replaySegment(s, visitor)) {
                 return;
             }
         }
@@ -307,19 +335,54 @@ private void trimSealedSegments(long ackedSeq) {
                 continue;
             }
             if (s.lastSeq() <= ackedSeq) {
+                // Close the fd up front: even if remove fails and the segment
+                // stays in the list, we won't read from it again — replay()
+                // skips removePending segments and append() never targets a
+                // sealed one. Holding the fd would just leak a descriptor.
                 if (s.fd != -1) {
                     ff.close(s.fd);
                     s.fd = -1;
                 }
+                boolean removed;
                 if (s.pathPtrNative != 0) {
-                    ff.remove(s.pathPtrNative);
-                    ff.freeNativePath(s.pathPtrNative);
-                    s.pathPtrNative = 0;
+                    removed = ff.remove(s.pathPtrNative);
                 } else {
                     // Recovery case: rotate's allocNativePath OOMed and left
                     // pathPtrNative=0. Fall back to the String form, which
                     // does its own one-shot encode/free internally.
-                    ff.remove(s.path);
+                    removed = ff.remove(s.path);
+                }
+                if (!removed) {
+                    // remove() failed (Windows sharing-violation under AV,
+                    // transient NFS error, ESTALE, etc.). DO NOT decrement
+                    // bytesOnDiskCache or free pathPtrNative — the file is
+                    // still on disk. Keep the segment in the in-memory list
+                    // so:
+                    //   (a) bytesOnDisk() keeps reporting the truth and the
+                    //       sf_max_total_bytes cap stays enforceable;
+                    //   (b) the next trim() retries the remove (the
+                    //       lastSeq() <= ackedSeq predicate still holds for
+                    //       cumulative ACKs);
+                    //   (c) replay() skips it via the removePending flag so
+                    //       already-acked frames don't re-ship to the new
+                    //       server on reconnect.
+                    if (!s.removePending) {
+                        LOG.warn("trim: remove() failed for sealed segment, "
+                                + "will retry on next trim [path={}, baseSeq={}, "
+                                + "lastSeq={}, writePos={}]",
+                                s.path, s.baseSeq, s.lastSeq(), s.writePos);
+                    }
+                    s.removePending = true;
+                    segments.setQuick(writeIdx++, s);
+                    continue;
+                }
+                if (s.removePending) {
+                    LOG.info("trim: retry succeeded for previously-failed "
+                            + "remove [path={}, baseSeq={}]", s.path, s.baseSeq);
+                }
+                if (s.pathPtrNative != 0) {
+                    ff.freeNativePath(s.pathPtrNative);
+                    s.pathPtrNative = 0;
                 }
                 bytesOnDiskCache -= s.writePos;
             } else {
@@ -373,6 +436,25 @@ public void close() {
                 ff.close(s.fd);
                 s.fd = -1;
             }
+            // Last-chance retry for segments whose mid-session remove() failed
+            // (e.g. Windows sharing-violation that has since cleared, transient
+            // NFS error that has resolved). Without this, an orphan .sfs file
+            // would persist on disk and the next process start would
+            // rediscover it via scanDirectory and replay its already-acked
+            // frames to the new server — silent duplicate writes.
+            if (s.removePending) {
+                boolean removed = s.pathPtrNative != 0
+                        ? ff.remove(s.pathPtrNative)
+                        : ff.remove(s.path);
+                if (removed) {
+                    s.removePending = false;
+                } else {
+                    LOG.warn("close: remove() still failing for orphaned segment "
+                            + "[path={}, baseSeq={}] — file will be rediscovered "
+                            + "on next start and re-replay its already-acked "
+                            + "frames to the new server", s.path, s.baseSeq);
+                }
+            }
             if (s.pathPtrNative != 0) {
                 ff.freeNativePath(s.pathPtrNative);
                 s.pathPtrNative = 0;
@@ -850,6 +932,15 @@ static final class Segment {
         long pathPtrNative;
         int fd = -1;
         boolean sealed;
+        // Trim attempted to delete this segment but ff.remove returned false
+        // (e.g. Windows sharing-violation, transient NFS error). The .sfs
+        // file is still on disk; the next trim() will retry the remove.
+        // While true, the segment stays in the in-memory list so:
+        //   (a) bytesOnDisk() keeps counting it (sf_max_total_bytes stays
+        //       enforceable),
+        //   (b) replay() skips it (its frames were already acked — must not
+        //       re-ship to the new server).
+        boolean removePending;
 
         long lastSeq() {
             return sealed ? lastSeqOnDisk : (baseSeq + frameCount - 1);