fix(ilp): cursor SF review fixes — perf, cap, torn-tail, races

Multi-issue review pass over the cursor SF stack from PR-17. Each
fix lands with a regression test (or extends an existing one); full
QWP suite green at 1053 tests.

- recovery sort: SegmentRing.openExisting used selection sort, which
  regressed the legacy SegmentLog perf fix (commit 86b6e6f) on the
  cursor path. Replaced with median-of-three in-place quicksort —
  median-of-three needed because lexicographic readdir returns our
  zero-padded hex names already-sorted, the worst case for a naive
  pivot. testLargeSegmentCountReopensInOrder guards against
  regression.

- cap accounting: SegmentManager.register/deregister ignored the
  bytes a recovered or adopted ring already owns, so the documented
  sf_max_total_bytes cap could be exceeded by a full segment-set on
  restart. Now seeds totalBytes from ring.totalSegmentBytes() and
  serviceRing updates the counter under the same lock so concurrent
  register/deregister stays consistent.

- torn-tail observability: openExisting silently truncated on the
  first bad CRC with no log or signal. Now distinguishes attempted-
  but-failed frame writes (non-zero bytes past lastGood) from clean
  unwritten space; emits WARN with byte count + offset and exposes
  tornTailBytes() for diagnostics. No false alarms on fresh hot
  spares or partial fills.

- drainOrphans Javadoc: setter claimed scan/log only with drainer
  runtime "as a follow-up", but build() spawns the
  BackgroundDrainerPool. Rewritten to describe actual adoption.

- BackgroundDrainerPool submit/close race: volatile-boolean gate
  let close() return while submit() was about to register a drainer.
  AtomicInteger CAS gate with sign-bit closed flag closes the window.

- CursorWebSocketSendLoop close hangs and reconnect leak: start()
  try/catches Thread.start() and assigns ioThread only on success;
  close() guards on isAlive() and closes the current client field;
  reconnect closes the previous WebSocketClient before assigning the
  new one.

- row-API methods accepting rows after the I/O loop went terminal:
  table()/longColumn()/atNow() etc. now surface the loop's terminal
  error on the next call instead of swallowing it until the next
  flush().

- non-success ACK not marking the loop fatal: the I/O response
  handler now routes through recordFatal so the loop exits cleanly
  instead of looping on the same rejected frame.

- CursorSendEngine close() left .sfa files when fully drained:
  captures fullyDrained before ring.close() and unlinks all segment
  files via unlinkAllSegmentFiles(dir) when nothing is pending
  replay.

- SegmentRing recovery leaked empty .sfa files: openExisting now
  Files.remove(path) after closing zero-frame segments instead of
  letting them accumulate across restart cycles.

- closeFlushTimeoutMillis sentinel collision: -1 was both the
  documented "no timeout" value and the parameter-not-set sentinel.
  Introduced CLOSE_FLUSH_TIMEOUT_NOT_SET = Long.MIN_VALUE; -1 now
  correctly disables the timeout.

- Windows files.c read/write/append accepted negative lengths and
  silently overflowed DWORD. New clamp_len helper rejects < 0 and
  clamps to MAXDWORD per call.

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

Author: bluestreak01

core/src/main/c/windows/files.c

@@ -145,14 +145,34 @@ JNIEXPORT jint JNICALL Java_io_questdb_client_std_Files_openCleanRW0
     return fd;
 }
 
+/* ReadFile/WriteFile take a DWORD (uint32) byte count, but the JNI signature
+ * exposes a jlong. A direct (DWORD) cast silently truncates the high 32 bits,
+ * which means a 4 GiB request becomes a 0-byte transfer — the worst kind of
+ * silent failure. Clamp to MAXDWORD so any oversized request is served as a
+ * short transfer (matching POSIX semantics on the share/files.c side); the
+ * Java caller already loops on the return value. Reject negative len up front
+ * so it doesn't get reinterpreted as a huge unsigned DWORD. */
+static inline DWORD clamp_len(jlong len) {
+    if (len > (jlong) MAXDWORD) {
+        return MAXDWORD;
+    }
+    return (DWORD) len;
+}
+
 JNIEXPORT jlong JNICALL Java_io_questdb_client_std_Files_read
         (JNIEnv *e, jclass cl, jint fd, jlong addr, jlong len, jlong offset) {
+    if (len < 0) {
+        SetLastError(ERROR_INVALID_PARAMETER);
+        SaveLastError();
+        return -1;
+    }
+    if (len == 0) return 0;
     OVERLAPPED ov;
     memset(&ov, 0, sizeof(ov));
     ov.Offset = (DWORD) (offset & 0xFFFFFFFF);
     ov.OffsetHigh = (DWORD) (offset >> 32);
     DWORD got = 0;
-    if (!ReadFile(FD_TO_HANDLE(fd), (LPVOID) (uintptr_t) addr, (DWORD) len, &got, &ov)) {
+    if (!ReadFile(FD_TO_HANDLE(fd), (LPVOID) (uintptr_t) addr, clamp_len(len), &got, &ov)) {
         DWORD err = GetLastError();
         if (err == ERROR_HANDLE_EOF) {
             return 0;
@@ -165,12 +185,18 @@ JNIEXPORT jlong JNICALL Java_io_questdb_client_std_Files_read
 
 JNIEXPORT jlong JNICALL Java_io_questdb_client_std_Files_write
         (JNIEnv *e, jclass cl, jint fd, jlong addr, jlong len, jlong offset) {
+    if (len < 0) {
+        SetLastError(ERROR_INVALID_PARAMETER);
+        SaveLastError();
+        return -1;
+    }
+    if (len == 0) return 0;
     OVERLAPPED ov;
     memset(&ov, 0, sizeof(ov));
     ov.Offset = (DWORD) (offset & 0xFFFFFFFF);
     ov.OffsetHigh = (DWORD) (offset >> 32);
     DWORD wrote = 0;
-    if (!WriteFile(FD_TO_HANDLE(fd), (LPCVOID) (uintptr_t) addr, (DWORD) len, &wrote, &ov)) {
+    if (!WriteFile(FD_TO_HANDLE(fd), (LPCVOID) (uintptr_t) addr, clamp_len(len), &wrote, &ov)) {
         SaveLastError();
         return -1;
     }
@@ -179,8 +205,14 @@ JNIEXPORT jlong JNICALL Java_io_questdb_client_std_Files_write
 
 JNIEXPORT jlong JNICALL Java_io_questdb_client_std_Files_append
         (JNIEnv *e, jclass cl, jint fd, jlong addr, jlong len) {
+    if (len < 0) {
+        SetLastError(ERROR_INVALID_PARAMETER);
+        SaveLastError();
+        return -1;
+    }
+    if (len == 0) return 0;
     DWORD wrote = 0;
-    if (!WriteFile(FD_TO_HANDLE(fd), (LPCVOID) (uintptr_t) addr, (DWORD) len, &wrote, NULL)) {
+    if (!WriteFile(FD_TO_HANDLE(fd), (LPCVOID) (uintptr_t) addr, clamp_len(len), &wrote, NULL)) {
         SaveLastError();
         return -1;
     }

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

@@ -681,8 +681,11 @@ public int getTimeout() {
         private SfDurability sfDurability = SfDurability.MEMORY;
         // close() drain timeout. Default applied at build() time. 0 or -1
         // means "fast close" (skip the drain entirely); any positive value
-        // bounds the wait for ackedFsn to catch up to publishedFsn.
-        private long closeFlushTimeoutMillis = PARAMETER_NOT_SET_EXPLICITLY;
+        // bounds the wait for ackedFsn to catch up to publishedFsn. Uses
+        // its own sentinel because -1 is a documented user-facing value
+        // and would otherwise collide with PARAMETER_NOT_SET_EXPLICITLY.
+        private static final long CLOSE_FLUSH_TIMEOUT_NOT_SET = Long.MIN_VALUE;
+        private long closeFlushTimeoutMillis = CLOSE_FLUSH_TIMEOUT_NOT_SET;
         // Reconnect policy. Defaults applied at build() time. Per-outage
         // time cap (default 300_000), initial backoff (default 100), and
         // max backoff (default 5_000) for the cursor I/O loop's exponential
@@ -1036,7 +1039,7 @@ public Sender build() {
                 long actualSfMaxTotalBytes = sfMaxTotalBytes == PARAMETER_NOT_SET_EXPLICITLY
                         ? Math.max(defaultMaxTotal, actualSfMaxBytes * 2)
                         : sfMaxTotalBytes;
-                long actualCloseFlushTimeoutMillis = closeFlushTimeoutMillis == PARAMETER_NOT_SET_EXPLICITLY
+                long actualCloseFlushTimeoutMillis = closeFlushTimeoutMillis == CLOSE_FLUSH_TIMEOUT_NOT_SET
                         ? DEFAULT_CLOSE_FLUSH_TIMEOUT_MILLIS
                         : closeFlushTimeoutMillis;
                 long actualReconnectMaxDurationMillis =
@@ -1919,20 +1922,22 @@ public LineSenderBuilder sfAppendDeadlineMillis(long millis) {
         }
 
         /**
-         * Opt in to scanning {@code <sf_dir>/*} at startup for sibling slots
-         * that hold unacked data left behind by a crashed sender or a
-         * different sender_id. Default {@code false}. WebSocket only;
+         * Opt in to adopting sibling slots under {@code <sf_dir>/*} at
+         * startup that hold unacked data left behind by a crashed sender or
+         * a different sender_id. Default {@code false}. WebSocket only;
          * requires {@code sf_dir} to be set.
          * <p>
-         * The scan is read-only — slots flagged with the {@code .failed}
-         * sentinel are skipped (manual reset required), and the foreground
-         * sender's own slot is never reported.
+         * On startup, after the foreground sender has acquired its own slot
+         * lock, the scan walks every sibling slot directory and dispatches a
+         * background drainer for each candidate orphan. Each drainer takes
+         * the slot's exclusive lock, replays the slot's unacked frames over
+         * its own WebSocket connection to the same target, and unlinks the
+         * slot once fully drained. Concurrency is capped by
+         * {@link #maxBackgroundDrainers(int)} (default {@code 4}).
          * <p>
-         * <b>Status:</b> the scan + visibility (via logs) lands in this
-         * release; the background drainer runtime that actually empties
-         * orphan slots is a follow-up. Setting {@code drain_orphans=true}
-         * today logs the count and paths of orphans found at startup so
-         * users can monitor + manually drain pending slots.
+         * Slots flagged with the {@code .failed} sentinel are skipped
+         * (manual reset required), and the foreground sender's own slot is
+         * never adopted.
          */
         public LineSenderBuilder drainOrphans(boolean enabled) {
             if (protocol != PARAMETER_NOT_SET_EXPLICITLY && protocol != PROTOCOL_WEBSOCKET) {

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

@@ -879,17 +879,30 @@ public QwpWebSocketSender floatColumn(CharSequence columnName, float value) {
     }
 
     /**
-     * Flushes buffered rows and waits until the server acknowledges all submitted
-     * WebSocket batches.
+     * Encodes pending rows into the cursor engine and returns once the data
+     * is published into the engine — in-RAM for memory mode, on-disk for
+     * store-and-forward mode. {@code flush()} does <b>not</b> wait for the
+     * server to acknowledge the batches; ACKs arrive asynchronously and the
+     * background I/O loop trims acked frames out of the engine independently.
      * <p>
-     * If a WebSocket send, receive, ACK timeout, server error ACK, invalid ACK,
-     * or server close is observed after the connection has been established, the
-     * sender enters a terminal failed state. The first failure is retained and
-     * subsequent public operations rethrow the same {@link LineSenderException}.
-     * Create a new sender to resume sending.
+     * If the engine's cursor ring is at the {@code sf_max_total_bytes} cap,
+     * {@code flush()} blocks while the I/O loop drains acked frames and
+     * frees space, up to {@code sf_append_deadline_millis} (default 30 s);
+     * on deadline expiry, this method throws.
+     * <p>
+     * For close-time drain semantics — waiting for the server to ACK
+     * everything published before shutting the I/O loop down — use
+     * {@link io.questdb.client.Sender.LineSenderBuilder#closeFlushTimeoutMillis(long)}.
+     * <p>
+     * If a WebSocket send, receive, ACK timeout, server error ACK, invalid
+     * ACK, or server close is observed after the connection has been
+     * established, the sender enters a terminal failed state. The first
+     * failure is retained and subsequent public operations rethrow the same
+     * {@link LineSenderException}. Create a new sender to resume sending.
      *
-     * @throws LineSenderException if the sender is closed, a row is still in
-     *                             progress, connection setup fails, or a terminal
+     * @throws LineSenderException if the sender is closed, a row is still
+     *                             in progress, connection setup fails, the
+     *                             engine cap deadline expires, or a terminal
      *                             WebSocket failure is observed
      */
     @Override
@@ -1445,6 +1458,16 @@ private void checkConnectionError() {
             error.fillInStackTrace();
             throw error;
         }
+        // Poll the cursor I/O loop's lastError too. Without this, a fatal
+        // wire / server-rejection error recorded by the I/O thread would
+        // only surface on the next flush() / close() — every row-level
+        // method (table, longColumn, atNow, etc.) routes through
+        // checkNotClosed → checkConnectionError, so failing to poll here
+        // means callers can keep accumulating rows long after the sender
+        // is already broken.
+        if (cursorSendLoop != null) {
+            cursorSendLoop.checkError();
+        }
     }
 
     private void checkTableSelected() {

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

@@ -32,6 +32,7 @@
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
 
 /**
  * Bounded thread pool that runs {@link BackgroundDrainer} tasks. One pool
@@ -51,11 +52,24 @@ public final class BackgroundDrainerPool implements QuietCloseable {
 
     private static final Logger LOG = LoggerFactory.getLogger(BackgroundDrainerPool.class);
     private static final long CLOSE_GRACE_MILLIS = 3_000L;
+    // CAS gate. Single AtomicInteger packs the closed flag (sign bit) and
+    // the in-flight submit count (low 31 bits):
+    //   state >= 0       → open, value is the in-flight submit count
+    //   state < 0        → closed bit set, low bits still track in-flight
+    //                      count waiting to drain
+    // submit() CASes state+1 only if state >= 0; close() CASes the CLOSED
+    // bit on, then waits for state to reach exactly CLOSED_BIT (no
+    // in-flight). This eliminates the "submit reads closed=false then
+    // close shuts the executor down" race window: the closed-bit CAS
+    // contends with the increment CAS on the same atomic, so submit
+    // either lands before close (and close waits for it to finish) or
+    // sees the closed bit and throws.
+    private static final int CLOSED_BIT = Integer.MIN_VALUE;
+    private final AtomicInteger state = new AtomicInteger();
 
     private final ExecutorService executor;
     private final CopyOnWriteArrayList<BackgroundDrainer> active = new CopyOnWriteArrayList<>();
     private final int maxConcurrent;
-    private volatile boolean closed;
 
     public BackgroundDrainerPool(int maxConcurrent) {
         if (maxConcurrent <= 0) {
@@ -77,19 +91,42 @@ public int maxConcurrent() {
      * Submits a drainer for background execution. The pool tracks it so
      * {@link #close} can request a stop. Safe to call any number of
      * times; excess submissions queue inside the pool's executor.
+     * <p>
+     * Reserves a "submit slot" on the {@link #state} CAS gate first; if
+     * the closed bit is already set, throws immediately. Otherwise the
+     * gate guarantees {@code close()} cannot shut the executor down until
+     * after we release the slot, so {@code executor.submit} always lands.
      */
     public void submit(BackgroundDrainer drainer) {
-        if (closed) {
-            throw new IllegalStateException("pool closed");
+        // Reserve a slot on the gate. Spin on CAS until either we win
+        // (state was non-negative) or we observe the closed bit.
+        for (;;) {
+            int s = state.get();
+            if (s < 0) {
+                throw new IllegalStateException("pool closed");
+            }
+            if (state.compareAndSet(s, s + 1)) break;
         }
-        active.add(drainer);
-        executor.submit(() -> {
-            try {
-                drainer.run();
-            } finally {
+        boolean accepted = false;
+        try {
+            active.add(drainer);
+            executor.submit(() -> {
+                try {
+                    drainer.run();
+                } finally {
+                    active.remove(drainer);
+                }
+            });
+            accepted = true;
+        } finally {
+            if (!accepted) {
                 active.remove(drainer);
             }
-        });
+            // Release our slot. Decrement is safe regardless of the
+            // closed bit's state — the bit lives in position 31 and
+            // only the low 31 bits move.
+            state.decrementAndGet();
+        }
     }
 
     /**
@@ -103,8 +140,21 @@ public java.util.List<BackgroundDrainer> snapshot() {
 
     @Override
     public void close() {
-        if (closed) return;
-        closed = true;
+        // Set the closed bit. CAS-loop because the in-flight count can be
+        // changing under us. Subsequent submit() calls will fail their
+        // CAS check (state < 0) and throw.
+        for (;;) {
+            int s = state.get();
+            if (s < 0) return; // already closed (idempotent)
+            if (state.compareAndSet(s, s | CLOSED_BIT)) break;
+        }
+        // Wait for in-flight submits to release their slots — i.e. for
+        // state to drain to exactly CLOSED_BIT (no low bits set). This
+        // ensures every submit's executor.submit has already returned
+        // before we shut the executor down.
+        while (state.get() != CLOSED_BIT) {
+            Thread.onSpinWait();
+        }
         for (BackgroundDrainer d : active) {
             d.requestStop();
         }