fix(ilp): harden SF disk-full retry, reconnect, and CRC paths

Multiple correctness fixes to the QWiP store-and-forward client that
were silently losing data under realistic outage scenarios.

Critical:
- markSending() moved after segmentLog.append; nextBatchSequence only
  advances on append success. Disk-full retry no longer crashes with
  IllegalStateException + drift exception, recycling the buffer
  without persistence (C1, C2).
- doReconnectCycle no longer drops pendingBuffer on every reconnect
  attempt. Buffer survives across attempts and is persisted by the
  post-reconnect ACTIVE state (C3).
- createActive closes fd in try/catch on writeHeader/fsync failure;
  no more fd leak on every failed rotation under disk pressure (C4).
- scanActive/replaySegment reject Files.length(fd) == -1 instead of
  treating it as "empty segment" (C5).

Moderate:
- scanActive distinguishes torn tail from mid-stream CRC mismatch;
  bit-rot followed by trailing bytes throws instead of silently
  truncating (M1).
- Files.close accepts any fd >= 0 (was refusing 0/1/2, leaking lock
  fd in containers where stdin/stdout/stderr were pre-closed) (M2).
- Connect-string sf_max_bytes / sf_max_total_bytes parsed as long;
  was capped at ~2 GB by parseIntValue (M3).
- WebSocketSendQueue.client made volatile so close-during-reconnect
  reads the live ref, not a stale one (M4).
- SegmentLog uses ObjList instead of java.util.ArrayList; bytesOnDisk
  is cached and updated incrementally so append() is O(1) zero-alloc
  on the I/O hot path (M6, N3).
- Each Segment caches a native UTF-8 path pointer; remove(String) is
  no longer called per-trim, eliminating the byte[] alloc on the I/O
  thread per ACK (M7).
- retryStalled always re-flags interrupt status (M8).

Cleanup:
- Dead WebSocketSendQueue.safeSendBatch removed (N1).
- @FunctionalInterface on Reconnector (N2).
- Inline FQNs in QwpWebSocketSender / Sender replaced with imports (N6).
- setSegmentLog overload pair co-located with cleaner doc (N8).
- Javadoc added to Files.java public surface and Crc32c.update (N9).
- Single-arg failConnection overload removed; every call site is now
  explicit about fatal vs non-fatal (N10).

Infrastructure:
- New FilesFacade interface + DefaultFilesFacade impl in
  io.questdb.client.std. SegmentLog refactored to use the facade so
  tests can inject OS-level failures (short writes, fstat -1, fsync
  EIO) without filesystem-level tricks.

Tests:
- 12 new red regression tests for the bug fixes above (now green).
- 5 coverage-gap tests for previously-untested error paths (M9):
  unsupported version header, baseSeq mismatch, multi-active
  rejection, oldestSeq edge cases, short-write recovery via
  fault-injection.
- Full SF + Files suite: 70 tests pass.

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

Author: bluestreak01

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

@@ -36,6 +36,8 @@
 import io.questdb.client.cutlass.line.tcp.PlainTcpLineChannel;
 import io.questdb.client.cutlass.qwp.client.QwpUdpSender;
 import io.questdb.client.cutlass.qwp.client.QwpWebSocketSender;
+import io.questdb.client.cutlass.qwp.client.sf.SegmentLog;
+import io.questdb.client.cutlass.qwp.client.sf.SfDiskFullException;
 import io.questdb.client.impl.ConfStringParser;
 import io.questdb.client.network.NetworkFacade;
 import io.questdb.client.network.NetworkFacadeImpl;
@@ -931,7 +933,7 @@ public Sender build() {
                     );
                 }
 
-                io.questdb.client.cutlass.qwp.client.sf.SegmentLog segmentLog = null;
+                SegmentLog segmentLog = null;
                 if (storeAndForward) {
                     if (sfDir == null) {
                         throw new LineSenderException(
@@ -942,12 +944,12 @@ public Sender build() {
                                 "store_and_forward requires async mode (in_flight_window > 1)");
                     }
                     long actualSfMaxBytes = sfMaxBytes == PARAMETER_NOT_SET_EXPLICITLY
-                            ? io.questdb.client.cutlass.qwp.client.sf.SegmentLog.DEFAULT_MAX_BYTES_PER_SEGMENT
+                            ? SegmentLog.DEFAULT_MAX_BYTES_PER_SEGMENT
                             : sfMaxBytes;
                     long actualSfMaxTotalBytes = sfMaxTotalBytes == PARAMETER_NOT_SET_EXPLICITLY
-                            ? io.questdb.client.cutlass.qwp.client.sf.SegmentLog.DEFAULT_MAX_TOTAL_BYTES
+                            ? SegmentLog.DEFAULT_MAX_TOTAL_BYTES
                             : sfMaxTotalBytes;
-                    segmentLog = io.questdb.client.cutlass.qwp.client.sf.SegmentLog.open(
+                    segmentLog = SegmentLog.open(
                             sfDir, actualSfMaxBytes, actualSfMaxTotalBytes, sfFsync);
                 } else if (sfDir != null) {
                     throw new LineSenderException(
@@ -1589,7 +1591,7 @@ public LineSenderBuilder storeAndForwardDir(String dir) {
 
         /**
          * Maximum bytes per segment file before rotation. Defaults to
-         * {@link io.questdb.client.cutlass.qwp.client.sf.SegmentLog#DEFAULT_MAX_BYTES_PER_SEGMENT}
+         * {@link SegmentLog#DEFAULT_MAX_BYTES_PER_SEGMENT}
          * (64 MiB). Smaller segments mean faster trim of acked data; larger
          * segments mean fewer rotations.
          */
@@ -1606,7 +1608,7 @@ public LineSenderBuilder storeAndForwardMaxBytes(long maxBytes) {
 
         /**
          * Hard cap on total bytes consumed by SF on disk. When the cap is reached,
-         * subsequent appends throw {@link io.questdb.client.cutlass.qwp.client.sf.SfDiskFullException}
+         * subsequent appends throw {@link SfDiskFullException}
          * which propagates as back-pressure: {@code flush()} blocks on the user
          * thread until ACKs trim acknowledged segments and free space. Default is
          * unbounded ({@link Long#MAX_VALUE}).
@@ -1694,6 +1696,17 @@ private static int parseIntValue(@NotNull StringSink value, @NotNull String name
             }
         }
 
+        private static long parseLongValue(@NotNull StringSink value, @NotNull String name) {
+            if (Chars.isBlank(value)) {
+                throw new LineSenderException(name).put(" cannot be empty");
+            }
+            try {
+                return Numbers.parseLong(value);
+            } catch (NumericException e) {
+                throw new LineSenderException("invalid ").put(name).put(" [value=").put(value).put("]");
+            }
+        }
+
         private static int resolveIPv4(String host) {
             try {
                 byte[] addr = InetAddress.getByName(host).getAddress();
@@ -2071,14 +2084,14 @@ private LineSenderBuilder fromConfig(CharSequence configurationString) {
                         throw new LineSenderException("sf_max_bytes is only supported for WebSocket transport");
                     }
                     pos = getValue(configurationString, pos, sink, "sf_max_bytes");
-                    long maxBytes = parseIntValue(sink, "sf_max_bytes");
+                    long maxBytes = parseLongValue(sink, "sf_max_bytes");
                     storeAndForwardMaxBytes(maxBytes);
                 } else if (Chars.equals("sf_max_total_bytes", sink)) {
                     if (protocol != PROTOCOL_WEBSOCKET) {
                         throw new LineSenderException("sf_max_total_bytes is only supported for WebSocket transport");
                     }
                     pos = getValue(configurationString, pos, sink, "sf_max_total_bytes");
-                    long maxTotal = parseIntValue(sink, "sf_max_total_bytes");
+                    long maxTotal = parseLongValue(sink, "sf_max_total_bytes");
                     storeAndForwardMaxTotalBytes(maxTotal);
                 } else if (Chars.equals("sf_fsync", sink)) {
                     if (protocol != PROTOCOL_WEBSOCKET) {

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

@@ -33,6 +33,7 @@
 import io.questdb.client.cutlass.line.LineSenderException;
 import io.questdb.client.cutlass.line.array.DoubleArray;
 import io.questdb.client.cutlass.line.array.LongArray;
+import io.questdb.client.cutlass.qwp.client.sf.SegmentLog;
 import io.questdb.client.cutlass.qwp.protocol.QwpConstants;
 import io.questdb.client.cutlass.qwp.protocol.QwpTableBuffer;
 import io.questdb.client.std.CharSequenceLongHashMap;
@@ -167,7 +168,7 @@ public class QwpWebSocketSender implements Sender {
     private boolean sawBinaryAck;
     private boolean sawPong;
     private WebSocketSendQueue sendQueue;
-    private io.questdb.client.cutlass.qwp.client.sf.SegmentLog segmentLog;
+    private SegmentLog segmentLog;
     // True when this sender took ownership of segmentLog (e.g. via the
     // connect-string builder); close() will then close the log too.
     private boolean ownsSegmentLog;
@@ -348,7 +349,7 @@ public static QwpWebSocketSender connect(
             String authorizationHeader,
             int maxSchemasPerConnection,
             boolean requestDurableAck,
-            io.questdb.client.cutlass.qwp.client.sf.SegmentLog segmentLog
+            SegmentLog segmentLog
     ) {
         QwpWebSocketSender sender = new QwpWebSocketSender(
                 host, port, tlsConfig,
@@ -1192,38 +1193,32 @@ public void setRequestDurableAck(boolean enabled) {
     }
 
     /**
-     * Attach a store-and-forward log. Every outgoing batch is captured to disk
-     * before the wire send and trimmed on cumulative ACK; the log also becomes
-     * the batch-sequence authority so sequencing survives sender restarts. The
-     * caller retains ownership of the log and is responsible for closing it
-     * after this sender has been closed.
+     * Attach a store-and-forward log to capture every outgoing batch to disk
+     * before the wire send and trim it on cumulative ACK. The log also becomes
+     * the batch-sequence authority so sequencing survives sender restarts.
      * <p>
-     * Requires async mode ({@code inFlightWindowSize > 1}).
+     * The caller retains ownership of {@code log} and is responsible for
+     * closing it after this sender has been closed; use the two-arg overload
+     * {@link #setSegmentLog(SegmentLog, boolean)} to transfer ownership.
+     * <p>
+     * Must be called before the first send. Requires async mode
+     * ({@code inFlightWindowSize > 1}).
      *
      * @throws LineSenderException if the sender is already connected or closed,
      *                             or if async mode is not enabled
      */
-    public void setSegmentLog(io.questdb.client.cutlass.qwp.client.sf.SegmentLog log) {
+    public void setSegmentLog(SegmentLog log) {
         setSegmentLog(log, false);
     }
 
     /**
-     * Number of times an outgoing batch was stalled because the SF total disk cap
-     * was reached. Each stall blocks the user thread's flush() until ACKs trim
-     * sealed segments and free space. Useful for monitoring backpressure under
-     * production load.
-     */
-    public long getTotalSfDiskFullStalls() {
-        return sendQueue == null ? 0 : sendQueue.getTotalDiskFullStalls();
-    }
-
-    /**
-     * Like {@link #setSegmentLog(io.questdb.client.cutlass.qwp.client.sf.SegmentLog)} but
-     * with explicit ownership transfer: when {@code takeOwnership} is true, this
-     * sender will close the log on its own {@link #close()}. Used by the
-     * connect-string builder to give the sender a self-contained lifecycle.
+     * Like {@link #setSegmentLog(SegmentLog)} but with explicit ownership
+     * transfer: when {@code takeOwnership} is true the sender closes
+     * {@code log} on its own {@link #close()}. Used by the connect-string
+     * builder to give the sender a self-contained lifecycle. Pass
+     * {@code false} to keep ownership with the caller.
      */
-    public void setSegmentLog(io.questdb.client.cutlass.qwp.client.sf.SegmentLog log, boolean takeOwnership) {
+    public void setSegmentLog(SegmentLog log, boolean takeOwnership) {
         if (closed) {
             throw new LineSenderException("Sender is closed");
         }
@@ -1239,6 +1234,16 @@ public void setSegmentLog(io.questdb.client.cutlass.qwp.client.sf.SegmentLog log
         this.ownsSegmentLog = takeOwnership && log != null;
     }
 
+    /**
+     * Number of times an outgoing batch was stalled because the SF total disk cap
+     * was reached. Each stall blocks the user thread's flush() until ACKs trim
+     * sealed segments and free space. Useful for monitoring backpressure under
+     * production load.
+     */
+    public long getTotalSfDiskFullStalls() {
+        return sendQueue == null ? 0 : sendQueue.getTotalDiskFullStalls();
+    }
+
     /**
      * Adds a SHORT column value to the current row.
      *

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

@@ -44,6 +44,7 @@
  * retry. Connection-fatal errors (auth failure, protocol mismatch) should still
  * be thrown; classification of fatal vs recoverable is the caller's job.
  */
+@FunctionalInterface
 public interface Reconnector {
     WebSocketClient reconnect() throws Exception;
 }