fix(ilp): cursor SF — apply PR-17 review critical and moderate findings

Critical:
- C1: SegmentRing.openExisting quarantines corrupt-first-frame .sfa
  files to <path>.corrupt instead of unlinking, preventing silent
  loss of valid frames behind a bit-flipped frame[0] CRC.
- C2: clamp wireSeq in CursorWebSocketSendLoop.handleServerRejection
  against nextWireSeq-1 (matches OK-ACK branch); SegmentRing.acknowledge
  also clamps at publishedFsn for defense-in-depth.
- C4: recordFatal now runs before dispatchError at all four HALT sites
  in CursorWebSocketSendLoop so the typed terminal error is latched
  before the user handler is invoked.
- C5: MmapSegment.create removes the on-disk file on mmap-fail catch;
  SegmentManager.serviceRing cleanup removes the path even when the
  spare is null.
- C7: remove stray QWP_CLIENT_REVIEW.md (review notes for vi_egress).
- C8: Sender.build wraps startOrphanDrainers in its own try/catch that
  closes the connected sender on failure; the outer catch (which
  closes cursorEngine directly) only fires for the pre-connect window
  before ownership transfer.

Moderate:
- M1: MmapSegment.openExisting validates baseSeq >= 0 and throws
  MmapSegmentException so SegmentRing's skip-with-log handles it.
- M2: SegmentRing.openExisting catches Throwable per-file and wraps
  the whole recovery body in an outer catch that closes every
  recovered segment on rethrow, plugging fd+mmap leaks.
- M3: POSIX read/write/append reject negative len with EINVAL,
  matching the existing Win32 guard.
- M4: crc32c.c _Static_assert on __BYTE_ORDER__ so big-endian builds
  fail loudly rather than silently miscompiling slice-by-8.

Tests:
- C9: ServerErrorAckTerminalTest and IoThreadErrorSurfacedOnRowApiTest
  flipped from STATUS_SCHEMA_MISMATCH (DROP) to STATUS_PARSE_ERROR
  (HALT) — the terminal-throw contract being asserted is the HALT
  contract per spec. New testDropPolicyNackDoesNotHaltAndAdvancesAck
  pins the DROP_AND_CONTINUE contract.
- C10: MmapSegmentTest.testFirstFrameCrcCorruptionFlagsTornTailAnd
  PreservesFile covers the unit-level contract for corrupt frame[0].
- C11: PrReviewRedTestsE2e adds end-to-end coverage for the central
  user-visible error API contract — flush() after a HALT NACK throws
  LineSenderServerException carrying the typed SenderError.
- SegmentRingTest.testAcknowledgeIsMonotonic publishes frames before
  acking to reflect the new clamp-at-publishedFsn contract.

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

Author: bluestreak01

QWP_CLIENT_REVIEW.md

@@ -25,6 +25,27 @@
 #include <jni.h>
 #include <stdint.h>
 #include <stddef.h>
+#include <assert.h>
+
+/*
+ * Slice-by-8 fold below assumes a little-endian byte order: the
+ * __builtin_memcpy of the first 4 bytes into a uint32_t is XORed against
+ * `crc` and then sliced as crc & 0xff / (crc >> 8) & 0xff / (crc >> 16) &
+ * 0xff / (crc >> 24) & 0xff. On big-endian (s390x, ppc64be) this would
+ * shift the bytes through the wrong tables and silently produce wrong
+ * CRCs — which on the SF path would manifest as data-loss-after-recovery
+ * because a bit-correct frame would still fail the integrity check.
+ *
+ * QuestDB's shipped binaries are all little-endian (linux/macOS x86_64
+ * and aarch64, Windows x86_64), so this is a forward-looking guard rather
+ * than a runtime fix. Using the static-assertion form failing the build
+ * is the right answer; we do not want a compile-time-best-effort fallback
+ * to a portable byte-by-byte path that miscompiles silently.
+ */
+#if defined(__BYTE_ORDER__) && defined(__ORDER_LITTLE_ENDIAN__)
+_Static_assert(__BYTE_ORDER__ == __ORDER_LITTLE_ENDIAN__,
+               "CRC32C slice-by-8 requires little-endian byte order");
+#endif
 
 /*
  * CRC-32C (Castagnoli) software implementation, reflected — slice-by-8.

core/src/main/c/share/crc32c.c

@@ -94,20 +94,37 @@ JNIEXPORT jint JNICALL Java_io_questdb_client_std_Files_openCleanRW0
 
 JNIEXPORT jlong JNICALL Java_io_questdb_client_std_Files_read
         (JNIEnv *e, jclass cl, jint fd, jlong addr, jlong len, jlong offset) {
+    // Reject negative len explicitly: jlong is signed but pread takes a
+    // size_t. Without this guard the cast wraps a small negative value
+    // into an enormous unsigned read length and the kernel may either
+    // SEGV on the address space or scribble far past the caller's buffer.
+    // The Win32 path already does this; matching here.
+    if (len < 0) {
+        errno = EINVAL;
+        return -1;
+    }
     ssize_t res;
     RESTARTABLE(pread((int) fd, (void *) (uintptr_t) addr, (size_t) len, (off_t) offset), res);
     return (jlong) res;
 }
 
 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) {
+        errno = EINVAL;
+        return -1;
+    }
     ssize_t res;
     RESTARTABLE(pwrite((int) fd, (const void *) (uintptr_t) addr, (size_t) len, (off_t) offset), res);
     return (jlong) res;
 }
 
 JNIEXPORT jlong JNICALL Java_io_questdb_client_std_Files_append
         (JNIEnv *e, jclass cl, jint fd, jlong addr, jlong len) {
+    if (len < 0) {
+        errno = EINVAL;
+        return -1;
+    }
     ssize_t res;
     RESTARTABLE(write((int) fd, (const void *) (uintptr_t) addr, (size_t) len), res);
     return (jlong) res;

core/src/main/c/share/files.c

@@ -1091,11 +1091,12 @@ public Sender build() {
                 CursorSendEngine cursorEngine = new CursorSendEngine(
                         slotPath, actualSfMaxBytes,
                         actualSfMaxTotalBytes, actualSfAppendDeadlineNanos);
+                int actualErrorInboxCapacity = errorInboxCapacity != PARAMETER_NOT_SET_EXPLICITLY
+                        ? errorInboxCapacity
+                        : io.questdb.client.cutlass.qwp.client.sf.cursor.SenderErrorDispatcher.DEFAULT_CAPACITY;
+                QwpWebSocketSender connected;
                 try {
-                    int actualErrorInboxCapacity = errorInboxCapacity != PARAMETER_NOT_SET_EXPLICITLY
-                            ? errorInboxCapacity
-                            : io.questdb.client.cutlass.qwp.client.sf.cursor.SenderErrorDispatcher.DEFAULT_CAPACITY;
-                    QwpWebSocketSender connected = QwpWebSocketSender.connect(
+                    connected = QwpWebSocketSender.connect(
                             hosts.getQuick(0),
                             ports.getQuick(0),
                             wsTlsConfig,
@@ -1115,6 +1116,24 @@ public Sender build() {
                             errorHandler,
                             actualErrorInboxCapacity
                     );
+                } catch (Throwable t) {
+                    // connect() failed before ownership of cursorEngine
+                    // transferred — close it ourselves.
+                    try {
+                        cursorEngine.close();
+                    } catch (Throwable ignored) {
+                        // best-effort
+                    }
+                    throw t;
+                }
+                // connect() succeeded — `connected` now owns cursorEngine
+                // via setCursorEngine(engine, true). From here on, ANY
+                // failure must close `connected` (which closes the engine
+                // through ownsCursorEngine), not cursorEngine directly:
+                // closing the engine alone would leak the I/O thread,
+                // dispatcher daemon, drainer pool, microbatch buffers and
+                // WebSocketClient inside the abandoned `connected`.
+                try {
                     // Once the foreground sender is up, dispatch drainers
                     // for any sibling orphan slots. Scan AFTER we acquire
                     // our own slot lock so we never accidentally try to
@@ -1139,7 +1158,7 @@ public Sender build() {
                     return connected;
                 } catch (Throwable t) {
                     try {
-                        cursorEngine.close();
+                        connected.close();
                     } catch (Throwable ignored) {
                         // best-effort
                     }

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

@@ -434,8 +434,13 @@ private void fail(Throwable initial) {
                             System.nanoTime()
                     );
                     totalServerErrors.incrementAndGet();
-                    dispatchError(err);
+                    // recordFatal MUST run before dispatchError: the spec
+                    // requires signal.terminalError to be latched BEFORE the
+                    // handler is invoked, so a handler that synchronously
+                    // probes getLastTerminalError() (or calls flush()) sees
+                    // the typed error rather than null.
                     recordFatal(new LineSenderServerException(err), err);
+                    dispatchError(err);
                     return;
                 }
                 lastReconnectError = e;
@@ -481,8 +486,10 @@ private void fail(Throwable initial) {
                 System.nanoTime()
         );
         totalServerErrors.incrementAndGet();
-        dispatchError(err);
+        // recordFatal MUST run before dispatchError so the producer-observable
+        // terminal error is latched before the handler is invoked.
         recordFatal(new LineSenderServerException(err), err);
+        dispatchError(err);
     }
 
     /**
@@ -788,8 +795,11 @@ public void onClose(int code, String reason) {
                         System.nanoTime()
                 );
                 totalServerErrors.incrementAndGet();
-                dispatchError(err);
+                // recordFatal MUST run before dispatchError so the producer-
+                // observable terminal error is latched before the handler is
+                // invoked.
                 recordFatal(new LineSenderServerException(err), err);
+                dispatchError(err);
                 return;
             }
             fail(new LineSenderException(
@@ -837,7 +847,18 @@ private void handleServerRejection(long wireSeq) {
             byte status = response.getStatus();
             SenderError.Category category = classify(status);
             SenderError.Policy policy = defaultPolicyFor(category);
-            long fsn = fsnAtZero + Math.max(0L, wireSeq);
+            // Same sanity clamp as the success branch above: do not trust a
+            // rejection wireSeq beyond what we've actually sent. Without this
+            // clamp the DROP path advances ackedFsn past publishedFsn, which
+            // makes the segment manager trim sealed segments the I/O thread
+            // is still reading — and the next Unsafe.getInt SEGVs the JVM.
+            long highestSent = nextWireSeq - 1L;
+            long cappedSeq = Math.max(0L, Math.min(wireSeq, highestSent));
+            if (cappedSeq < wireSeq) {
+                LOG.warn("server NACK wire seq {} exceeds highest sent {} — clamping",
+                        wireSeq, highestSent);
+            }
+            long fsn = fsnAtZero + cappedSeq;
             // Best-effort table attribution: the parser populates
             // response.tableNames on error frames the same way it does on
             // STATUS_OK. If exactly one table was named, surface it; if
@@ -857,27 +878,26 @@ private void handleServerRejection(long wireSeq) {
                     System.nanoTime()
             );
             totalServerErrors.incrementAndGet();
-            // Async-deliver to the user handler regardless of policy. HALT
-            // also surfaces synchronously via the producer-thread typed throw
-            // below; DROP is observable ONLY via the async path, so the
-            // dispatcher is the user's only chance to dead-letter the data.
-            dispatchError(err);
 
             if (policy == SenderError.Policy.HALT) {
-                // Terminal: stash the typed payload, raise a typed exception
-                // through the existing recordFatal -> checkError -> producer
-                // throw path. Bytes on disk are the bytes the server
-                // rejected; reconnect/replay cannot fix them.
+                // Terminal: stash the typed payload BEFORE dispatching to the
+                // handler. The spec requires signal.terminalError to be latched
+                // before the handler is invoked so a handler that synchronously
+                // probes getLastTerminalError() (or calls flush()) sees the
+                // typed error rather than null. Bytes on disk are the bytes
+                // the server rejected; reconnect/replay cannot fix them.
                 recordFatal(new LineSenderServerException(err), err);
+                dispatchError(err);
             } else {
                 // DROP_AND_CONTINUE: advance ackedFsn past the rejected span
                 // so the loop drains subsequent batches. The data is dropped
                 // from the SF disk store via the existing trim path; the
-                // dispatch above is the user's only handle to dead-letter.
+                // dispatch is the user's only handle to dead-letter.
                 LOG.warn("server rejected wire seq {} (category={}, status=0x{}) — dropping batch and continuing",
                         wireSeq, category, Integer.toHexString(status & 0xFF));
                 engine.acknowledge(fsn);
                 totalAcks.incrementAndGet();
+                dispatchError(err);
             }
         }
     }