fix(ilp): cross-platform drainer/slot-lock and quieter PARSE_ERROR logs

BackgroundDrainerPool.close() called requestStop() before shutdown(),
forcing in-flight drainers to exit as STOPPED instead of SUCCESS. Their
engine.close() then saw fullyDrained=false and skipped the .sfa cleanup,
so drain_orphans=true left orphan slots on disk on macOS where the race
went the other way. Now we shutdown() first and let drainers complete
naturally within a grace window; requestStop() is the fallback.

SlotLock kept the holder's PID inside the .lock file itself. Windows'
LockFileEx is a mandatory whole-file lock, so a contender reading the
PID got 0 bytes and surfaced "holder=unknown" instead of "pid=<n>".
Move the PID to a sibling .lock.pid sidecar that nobody locks — the
diagnostic now works uniformly on POSIX and Windows.

CursorWebSocketSendLoop.recordFatal() logged the full throwable for
server-side rejects like PARSE_ERROR. The structured info is already
in the message and the dispatcher's one-line log; suppress the stack
trace for serverError != null and keep it for genuine client-side
failures.

Author: bluestreak01

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

@@ -52,7 +52,13 @@
 public final class BackgroundDrainerPool implements QuietCloseable {
 
     private static final Logger LOG = LoggerFactory.getLogger(BackgroundDrainerPool.class);
-    private static final long CLOSE_GRACE_MILLIS = 3_000L;
+    // Time we let drainers finish their drain naturally before signaling
+    // stop. awaitTermination returns as soon as the last drainer exits,
+    // so this only matters when something is genuinely stuck.
+    private static final long GRACEFUL_DRAIN_MILLIS = 2_500L;
+    // After signaling stop, give drainers a brief window to unwind cleanly
+    // (release slot lock, close engine) before forcing shutdownNow.
+    private static final long STOP_GRACE_MILLIS = 500L;
     // 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
@@ -160,16 +166,25 @@ public void close() {
         while (state.get() != CLOSED_BIT) {
             Thread.onSpinWait();
         }
-        for (BackgroundDrainer d : active) {
-            d.requestStop();
-        }
+        // Reject new tasks but let in-flight drainers finish their drain
+        // naturally. Without this grace window a drainer that's seconds
+        // away from acked >= target gets requestStop()'d and exits as
+        // STOPPED — its engine.close() then sees fullyDrained=false and
+        // leaves the slot's .sfa files behind, defeating drain_orphans.
         executor.shutdown();
         try {
-            if (!executor.awaitTermination(CLOSE_GRACE_MILLIS, TimeUnit.MILLISECONDS)) {
-                LOG.warn("drainer pool did not finish in {}ms; "
-                                + "remaining drainers will exit on their own",
-                        CLOSE_GRACE_MILLIS);
-                executor.shutdownNow();
+            if (!executor.awaitTermination(GRACEFUL_DRAIN_MILLIS, TimeUnit.MILLISECONDS)) {
+                LOG.warn("orphan drainers still running after {}ms — signaling stop",
+                        GRACEFUL_DRAIN_MILLIS);
+                for (BackgroundDrainer d : active) {
+                    d.requestStop();
+                }
+                if (!executor.awaitTermination(STOP_GRACE_MILLIS, TimeUnit.MILLISECONDS)) {
+                    LOG.warn("drainer pool did not exit in {}ms after stop; "
+                                    + "remaining drainers will exit on their own",
+                            STOP_GRACE_MILLIS);
+                    executor.shutdownNow();
+                }
             }
         } catch (InterruptedException e) {
             Thread.currentThread().interrupt();

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

@@ -591,7 +591,11 @@ private void recordFatal(Throwable t, SenderError serverError) {
             lastTerminalServerError = serverError;
         }
         running = false;
-        LOG.error("Cursor I/O loop failure: {}", t.getMessage(), t);
+        if (serverError != null) {
+            LOG.error("Cursor I/O loop failure: {}", t.getMessage());
+        } else {
+            LOG.error("Cursor I/O loop failure: {}", t.getMessage(), t);
+        }
     }
 
     /**

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

@@ -34,15 +34,18 @@
 /**
  * Advisory exclusive lock for a single SF slot directory.
  * <p>
- * One {@code .lock} file per slot, held via {@code flock} for the entire
- * lifetime of the engine that owns the slot. The lock is automatically
- * released when the fd is closed — including on hard process exit, since
- * the kernel cleans up flocks for terminated processes (see flock(2)).
+ * One {@code .lock} file per slot, held via {@code flock}/{@code LockFileEx}
+ * for the entire lifetime of the engine that owns the slot. The lock is
+ * automatically released when the fd is closed — including on hard process
+ * exit, since the kernel cleans up file locks for terminated processes.
  * <p>
- * The lock file's payload is the holder's PID, written at acquisition
- * time. A failed acquisition reads it back so the error message can name
- * the offending process — turning a vague "slot in use" into actionable
- * diagnostics.
+ * The holder's PID is written to a sibling {@code .lock.pid} file at
+ * acquisition time. A failed acquisition reads it back so the error message
+ * can name the offending process — turning a vague "slot in use" into
+ * actionable diagnostics. The PID lives in a separate file because Windows'
+ * {@code LockFileEx} is a mandatory range lock: while the {@code .lock}
+ * file is held, a second handle cannot read its bytes, so we couldn't
+ * recover the holder's PID from the lock file itself.
  * <p>
  * Two senders pointing at the same slot dir is the multi-writer footgun
  * the slot model exists to prevent: their FSN sequences would interleave
@@ -53,6 +56,7 @@
 public final class SlotLock implements QuietCloseable {
 
     private static final String LOCK_FILE_NAME = ".lock";
+    private static final String LOCK_PID_FILE_NAME = ".lock.pid";
     private final String slotDir;
     private final String lockPath;
     private int fd;
@@ -83,6 +87,7 @@ public static SlotLock acquire(String slotDir) {
             }
         }
         String lockPath = slotDir + "/" + LOCK_FILE_NAME;
+        String pidPath = slotDir + "/" + LOCK_PID_FILE_NAME;
         int fd = Files.openRW(lockPath);
         if (fd < 0) {
             throw new IllegalStateException(
@@ -92,12 +97,12 @@ public static SlotLock acquire(String slotDir) {
         try {
             int rc = Files.lock(fd);
             if (rc != 0) {
-                String holder = readHolder(lockPath);
+                String holder = readHolder(pidPath);
                 throw new IllegalStateException(
                         "sf slot already in use by another process [slot="
                                 + slotDir + ", holder=" + holder + "]");
             }
-            writePid(fd);
+            writePid(pidPath);
             ok = true;
             return new SlotLock(slotDir, lockPath, fd);
         } finally {
@@ -114,17 +119,18 @@ public String slotDir() {
 
     @Override
     public void close() {
-        // Closing the fd releases the flock. We do NOT remove the file —
-        // a stale .lock with the previous PID is harmless (next acquirer
-        // can flock it just fine, and overwrites the PID on success).
+        // Closing the fd releases the lock. We do NOT remove the .lock
+        // file or the .lock.pid sidecar — a stale PID is harmless (next
+        // acquirer overwrites .lock.pid on success).
         if (fd >= 0) {
             Files.close(fd);
             fd = -1;
         }
     }
 
-    private static String readHolder(String lockPath) {
-        int rfd = Files.openRO(lockPath);
+    private static String readHolder(String pidPath) {
+        if (!Files.exists(pidPath)) return "unknown";
+        int rfd = Files.openRO(pidPath);
         if (rfd < 0) return "unknown";
         try {
             long fileLen = Files.length(rfd);
@@ -147,24 +153,33 @@ private static String readHolder(String lockPath) {
         }
     }
 
-    private static void writePid(int fd) {
+    private static void writePid(String pidPath) {
         long pid;
         try {
             pid = ProcessHandle.current().pid();
         } catch (Throwable ignored) {
             // Diagnostic-only — never block lock acquisition on it.
             pid = -1L;
         }
-        byte[] payload = (pid + "\n").getBytes(StandardCharsets.UTF_8);
-        Files.truncate(fd, 0L);
-        long addr = Unsafe.malloc(payload.length, MemoryTag.NATIVE_DEFAULT);
+        int wfd = Files.openRW(pidPath);
+        if (wfd < 0) {
+            // Diagnostic-only — never block lock acquisition on it.
+            return;
+        }
         try {
-            for (int i = 0; i < payload.length; i++) {
-                Unsafe.getUnsafe().putByte(addr + i, payload[i]);
+            Files.truncate(wfd, 0L);
+            byte[] payload = (pid + "\n").getBytes(StandardCharsets.UTF_8);
+            long addr = Unsafe.malloc(payload.length, MemoryTag.NATIVE_DEFAULT);
+            try {
+                for (int i = 0; i < payload.length; i++) {
+                    Unsafe.getUnsafe().putByte(addr + i, payload[i]);
+                }
+                Files.write(wfd, addr, payload.length, 0L);
+            } finally {
+                Unsafe.free(addr, payload.length, MemoryTag.NATIVE_DEFAULT);
             }
-            Files.write(fd, addr, payload.length, 0L);
         } finally {
-            Unsafe.free(addr, payload.length, MemoryTag.NATIVE_DEFAULT);
+            Files.close(wfd);
         }
     }
 }