fix(ilp): harden QWiP dispatcher, disk allocation, and config parsing

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

@@ -164,9 +164,16 @@ JNIEXPORT jboolean JNICALL Java_io_questdb_client_std_Files_allocate
     fst.fst_length = (off_t) size;
     fst.fst_bytesalloc = 0;
     if (fcntl((int) fd, F_PREALLOCATE, &fst) == -1) {
+        /* Contiguous allocation failed (e.g. fragmented filesystem); retry
+         * non-contiguous all-or-nothing. Only fall through to ftruncate when
+         * the filesystem doesn't support F_PREALLOCATE at all; real failures
+         * (notably ENOSPC) must surface so the caller doesn't end up with a
+         * sparse file that SIGBUSes on later mmap store (sf-client.md §6). */
         fst.fst_flags = F_ALLOCATEALL;
-        (void) fcntl((int) fd, F_PREALLOCATE, &fst);
-        /* if F_PREALLOCATE fails we still try ftruncate to set logical size */
+        if (fcntl((int) fd, F_PREALLOCATE, &fst) == -1
+            && errno != ENOTSUP && errno != EOPNOTSUPP) {
+            return JNI_FALSE;
+        }
     }
 #endif
     int res2;

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

@@ -620,6 +620,10 @@ final class LineSenderBuilder {
         private static final long DEFAULT_WS_AUTO_FLUSH_INTERVAL_NANOS = 100_000_000L; // 100ms
         private static final int DEFAULT_WS_AUTO_FLUSH_ROWS = 1_000;
         private static final int MIN_BUFFER_SIZE = AuthUtils.CHALLENGE_LEN + 1; // challenge size + 1;
+        // sf-client.md section 4.4: the inbox capacity must accommodate the
+        // distinct error categories in a bursty error stream so that drop-oldest
+        // does not erase the trailing distribution of categories.
+        private static final int MIN_ERROR_INBOX_CAPACITY = 16;
         // The PARAMETER_NOT_SET_EXPLICITLY constant is used to detect if a parameter was set explicitly in configuration parameters
         // where it matters. This is needed to detect invalid combinations of parameters. Why?
         // We want to fail-fast even when an explicitly configured options happens to be same value as the default value,
@@ -1864,15 +1868,20 @@ public LineSenderBuilder errorHandler(io.questdb.client.SenderErrorHandler handl
          *
          * <p>WebSocket transport only; setting on other transports throws.
          *
-         * @param capacity must be {@code >= 1}
+         * @param capacity must be {@code >= 16} (sf-client.md section 4.4).
+         *                 The floor exists because overflow drops the oldest
+         *                 entry and watermarks are monotonic, so the inbox
+         *                 must be wide enough to keep a useful trailing
+         *                 window of categories under bursty errors.
          * @return this instance for method chaining
          */
         public LineSenderBuilder errorInboxCapacity(int capacity) {
             if (protocol != PARAMETER_NOT_SET_EXPLICITLY && protocol != PROTOCOL_WEBSOCKET) {
                 throw new LineSenderException("error_inbox_capacity is only supported for WebSocket transport");
             }
-            if (capacity < 1) {
-                throw new LineSenderException("error_inbox_capacity must be >= 1, was " + capacity);
+            if (capacity < MIN_ERROR_INBOX_CAPACITY) {
+                throw new LineSenderException("error_inbox_capacity must be >= "
+                        + MIN_ERROR_INBOX_CAPACITY + ", was " + capacity);
             }
             this.errorInboxCapacity = capacity;
             return this;

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

@@ -293,7 +293,9 @@ private QwpQueryClient(String host, int port) {
      * <ul>
      *   <li>{@code addr=host[:port][,host[:port]...]} -- required. Comma-separated list of
      *       WebSocket endpoints; {@link #connect()} walks them in order and stops at the
-     *       first matching {@code target=}. Default port on each entry is {@value #DEFAULT_WS_PORT}.</li>
+     *       first matching {@code target=}. Default port on each entry is {@value #DEFAULT_WS_PORT}.
+     *       Per {@code failover.md} section 1, the comma form and repeated {@code addr=} keys
+     *       both accumulate into a single ordered list; empty entries are rejected.</li>
      *   <li>{@code target=any|primary|replica} -- endpoint filter applied against the role
      *       byte from the v2 {@code SERVER_INFO} frame. Default {@code any}. {@code primary}
      *       accepts {@code PRIMARY}, {@code PRIMARY_CATCHUP} and {@code STANDALONE}.</li>
@@ -359,7 +361,7 @@ public static QwpQueryClient fromConfig(CharSequence configurationString) {
                     "unsupported schema [schema=" + sink + ", supported-schemas=[ws, wss]]");
         }
 
-        List<Endpoint> parsedEndpoints = null;
+        List<Endpoint> parsedEndpoints = new ArrayList<>();
         String path = DEFAULT_ENDPOINT_PATH;
         String target = TARGET_ANY;
         Boolean failover = null;
@@ -398,7 +400,9 @@ public static QwpQueryClient fromConfig(CharSequence configurationString) {
             String value = sink.toString();
             switch (key) {
                 case "addr":
-                    parsedEndpoints = parseEndpointList(value);
+                    // failover.md §1: comma syntax and repeated addr= keys must
+                    // accumulate. parseEndpointList rejects empty entries.
+                    parsedEndpoints.addAll(parseEndpointList(value));
                     break;
                 case "target":
                     if (!TARGET_ANY.equals(value) && !TARGET_PRIMARY.equals(value) && !TARGET_REPLICA.equals(value)) {
@@ -551,7 +555,7 @@ public static QwpQueryClient fromConfig(CharSequence configurationString) {
                     throw new IllegalArgumentException("unknown configuration key: " + key);
             }
         }
-        if (parsedEndpoints == null || parsedEndpoints.isEmpty()) {
+        if (parsedEndpoints.isEmpty()) {
             throw new IllegalArgumentException("missing required key: addr");
         }
         boolean hasBasic = username != null || password != null;

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

@@ -125,6 +125,9 @@ public class QwpWebSocketSender implements Sender {
     private static final int DEFAULT_MICROBATCH_BUFFER_SIZE = 1024 * 1024; // 1MB
     private static final Logger LOG = LoggerFactory.getLogger(QwpWebSocketSender.class);
     private static final int MAX_TABLE_NAME_LENGTH = 127;
+    // sf-client.md section 4.4 floor: drop-oldest under bursts needs a wide
+    // enough window to preserve the trailing category distribution.
+    private static final int MIN_ERROR_INBOX_CAPACITY = 16;
     private static final String WRITE_PATH = "/write/v4";
     private final String authorizationHeader;
     private final int autoFlushBytes;
@@ -1725,10 +1728,13 @@ public void setErrorHandler(SenderErrorHandler handler) {
     /**
      * Configure the bounded inbox capacity used by the dispatcher. Must be
      * called before {@code connect()}; later changes have no effect.
+     * The minimum follows sf-client.md section 4.4: drop-oldest under bursts
+     * needs a wide enough window to preserve the trailing category distribution.
      */
     public void setErrorInboxCapacity(int capacity) {
-        if (capacity < 1) {
-            throw new IllegalArgumentException("errorInboxCapacity must be >= 1, was " + capacity);
+        if (capacity < MIN_ERROR_INBOX_CAPACITY) {
+            throw new IllegalArgumentException("errorInboxCapacity must be >= "
+                    + MIN_ERROR_INBOX_CAPACITY + ", was " + capacity);
         }
         this.errorInboxCapacity = capacity;
     }

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

@@ -376,8 +376,11 @@ public long appendOrFsn(long payloadAddr, int payloadLen) {
             hotSpare = null;
             // Fresh active just consumed the spare → ask the manager to start
             // making the next one immediately, before this segment fills.
+            // The flag is per-active and tracks whether the backup-signal
+            // branch has fired for the *current* active. Rotation installs a
+            // new active, so the flag resets here to re-arm the backup branch.
             // Plain field reset is safe (producer-only state).
-            wakeupRequestedForActive = true;
+            wakeupRequestedForActive = false;
             Runnable wakeup = managerWakeup;
             if (wakeup != null) {
                 wakeup.run();

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

@@ -30,8 +30,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.util.concurrent.ArrayBlockingQueue;
-import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingDeque;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicLong;
 
@@ -61,10 +60,10 @@ public final class SenderConnectionDispatcher implements QuietCloseable {
             SenderConnectionEvent.NO_ROUND_NUMBER,
             null, 0L);
     private final AtomicLong dropped = new AtomicLong();
-    // volatile so the user can swap the listener post-connect, mirroring
-    // SenderErrorDispatcher's setHandler contract.
-    private volatile SenderConnectionListener listener;
-    private final BlockingQueue<SenderConnectionEvent> inbox;
+    // Deque (not plain queue) so offer() can drop the head when the inbox is
+    // full, per the drop-oldest contract inherited from SenderErrorDispatcher
+    // (sf-client.md section 14.6).
+    private final LinkedBlockingDeque<SenderConnectionEvent> inbox;
     // First offer() that observes a null thread wins the race to spawn it.
     private final Object lock = new Object();
     private final String threadName;
@@ -75,6 +74,9 @@ public final class SenderConnectionDispatcher implements QuietCloseable {
     // double-checked first-null guard is a JMM race -- benign in practice
     // (the synchronized re-check covers double-start) but spec-incorrect.
     private volatile Thread dispatcherThread;
+    // volatile so the user can swap the listener post-connect, mirroring
+    // SenderErrorDispatcher's setHandler contract.
+    private volatile SenderConnectionListener listener;
 
     public SenderConnectionDispatcher(SenderConnectionListener listener) {
         this(listener, DEFAULT_CAPACITY, "qdb-sf-connection-dispatcher");
@@ -92,7 +94,7 @@ public SenderConnectionDispatcher(SenderConnectionListener listener, int capacit
             throw new IllegalArgumentException("capacity must be >= 1, was " + capacity);
         }
         this.listener = listener;
-        this.inbox = new ArrayBlockingQueue<>(capacity);
+        this.inbox = new LinkedBlockingDeque<>(capacity);
         this.threadName = threadName;
     }
 
@@ -103,7 +105,6 @@ public void close() {
                 return;
             }
             closed = true;
-            //noinspection ResultOfMethodCallIgnored
             inbox.offer(POISON);
             Thread t = dispatcherThread;
             if (t != null) {
@@ -128,9 +129,10 @@ public void close() {
     }
 
     /**
-     * Total events dropped via inbox-overflow since startup. Non-zero means the
-     * listener is slower than the event rate -- typically a symptom of a
-     * misbehaving listener. Reported by the sender for ops dashboards.
+     * Total events discarded by the drop-oldest overflow policy since startup.
+     * Non-zero means the listener is slower than the event rate -- typically
+     * a symptom of a misbehaving listener. Reported by the sender for ops
+     * dashboards.
      */
     public long getDroppedNotifications() {
         return dropped.get();
@@ -145,36 +147,48 @@ public long getTotalDelivered() {
     }
 
     /**
-     * Replace the user-supplied listener. Effective for the next delivery.
-     * Null reverts to the loud-not-silent default.
-     */
-    public void setListener(SenderConnectionListener listener) {
-        this.listener = listener != null ? listener : DefaultSenderConnectionListener.INSTANCE;
-    }
-
-    /**
-     * Non-blocking enqueue. Returns {@code true} if the event will be
-     * delivered to the listener (eventually, on the dispatcher thread).
-     * Returns {@code false} if the inbox was full or the dispatcher was
-     * closed -- caller's only obligation is to not block.
+     * Non-blocking enqueue. Always admits the new event unless the dispatcher
+     * is closed or {@code event} is null. Returns {@code true} when the new
+     * event is enqueued for delivery, {@code false} when rejected outright.
+     *
+     * <p>When the inbox is full, drops the oldest pending entry to make room
+     * (sf-client.md section 14.6, inherited contract) and bumps
+     * {@link #getDroppedNotifications()}. The newest entry is always retained
+     * because later events carry the most recent connection state.
      *
      * <p>Lazy-starts the dispatcher thread on the first successful offer.
      */
     public boolean offer(SenderConnectionEvent event) {
         if (closed || event == null) {
             return false;
         }
-        boolean accepted = inbox.offer(event);
-        if (!accepted) {
-            dropped.incrementAndGet();
-            return false;
+        // Drop-oldest overflow policy. The pollFirst()/offerLast() pair is
+        // not atomic with the consumer, but the consumer can only remove
+        // entries (never add). The retry loop converges in at most one extra
+        // iteration under SPSC; close()'s POISON enqueue widens the race
+        // briefly but the `closed` re-check exits cleanly.
+        while (!inbox.offerLast(event)) {
+            if (inbox.pollFirst() != null) {
+                dropped.incrementAndGet();
+            }
+            if (closed) {
+                return false;
+            }
         }
         if (dispatcherThread == null) {
             startDispatcherIfNeeded();
         }
         return true;
     }
 
+    /**
+     * Replace the user-supplied listener. Effective for the next delivery.
+     * Null reverts to the loud-not-silent default.
+     */
+    public void setListener(SenderConnectionListener listener) {
+        this.listener = listener != null ? listener : DefaultSenderConnectionListener.INSTANCE;
+    }
+
     private void dispatchLoop() {
         while (!closed || !inbox.isEmpty()) {
             SenderConnectionEvent ev;

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

@@ -30,8 +30,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.util.concurrent.ArrayBlockingQueue;
-import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingDeque;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;
@@ -48,11 +47,14 @@
  * the daemon dispatcher takes from the queue and invokes the handler.
  *
  * <h2>Backpressure</h2>
- * The queue is bounded ({@code capacity}, default 256). When full,
- * {@link #offer} returns {@code false} immediately and bumps
+ * The inbox is bounded ({@code capacity}, default 256). When full, {@link #offer}
+ * drops the oldest entry to admit the new one and bumps
  * {@link #getDroppedNotifications()}. The I/O thread does NOT spin or block.
  * A non-zero dropped count means the handler is too slow to keep up — visible
- * to operators via the sender's accessor.
+ * to operators via the sender's accessor. Drop-oldest is mandated by
+ * sf-client.md section 14.6: watermarks are monotonic, so the latest entry is
+ * always the most informative and drops compress information rather than
+ * lose it.
  *
  * <h2>Lifecycle</h2>
  * The dispatcher thread is started lazily on the first successful
@@ -88,7 +90,11 @@ public final class SenderErrorDispatcher implements QuietCloseable {
     // and reconfigurable apps install a new handler at any time without
     // tearing down the dispatcher thread.
     private volatile SenderErrorHandler handler;
-    private final BlockingQueue<SenderError> inbox;
+    // Deque (not plain queue) so offer() can drop the head when the inbox is
+    // full, per spec section 14.6. SPSC in steady state: the I/O thread is
+    // the sole producer, the dispatcher is the sole consumer; close() also
+    // enqueues POISON, but only once and under `lock`.
+    private final LinkedBlockingDeque<SenderError> inbox;
     // Threads are started lazily under this monitor; takes the same role as
     // SegmentManager.start() — first offer() that observes a null thread
     // wins the race to spawn it.
@@ -118,7 +124,7 @@ public SenderErrorDispatcher(SenderErrorHandler handler, int capacity, String th
             throw new IllegalArgumentException("capacity must be >= 1, was " + capacity);
         }
         this.handler = handler;
-        this.inbox = new ArrayBlockingQueue<>(capacity);
+        this.inbox = new LinkedBlockingDeque<>(capacity);
         this.threadName = threadName;
     }
 
@@ -170,10 +176,10 @@ public void close() {
     }
 
     /**
-     * Total errors delivered via inbox-overflow drop since startup. Non-zero
-     * means the user's handler is slower than the error rate — typically a
-     * symptom of a misbehaving handler or a misconfigured server. Reported by
-     * the sender for ops dashboards.
+     * Total errors discarded by the drop-oldest overflow policy since startup.
+     * Non-zero means the user's handler is slower than the error rate —
+     * typically a symptom of a misbehaving handler or a misconfigured server.
+     * Reported by the sender for ops dashboards.
      */
     public long getDroppedNotifications() {
         return dropped.get();
@@ -209,21 +215,33 @@ public void setHandler(SenderErrorHandler handler) {
     }
 
     /**
-     * Non-blocking enqueue. Returns {@code true} if the error will be
-     * delivered to the handler (eventually, on the dispatcher thread). Returns
-     * {@code false} if the inbox was full or the dispatcher was closed —
-     * caller's only obligation is to not block.
+     * Non-blocking enqueue. Always admits the new error unless the dispatcher
+     * is closed or {@code error} is null. Returns {@code true} when the new
+     * error is enqueued for delivery, {@code false} when rejected outright.
+     *
+     * <p>When the inbox is full, drops the oldest pending entry to make room
+     * (sf-client.md section 14.6) and bumps {@link #getDroppedNotifications()}.
+     * The newest entry is always retained because it carries the most recent
+     * watermark information.
      *
      * <p>Lazy-starts the dispatcher thread on the first successful offer.
      */
     public boolean offer(SenderError error) {
         if (closed || error == null) {
             return false;
         }
-        boolean accepted = inbox.offer(error);
-        if (!accepted) {
-            dropped.incrementAndGet();
-            return false;
+        // Drop-oldest overflow policy. The pollFirst()/offerLast() pair is
+        // not atomic with the consumer, but the consumer can only remove
+        // entries (never add). The retry loop converges in at most one extra
+        // iteration under SPSC; close()'s POISON enqueue widens the race
+        // briefly but the `closed` re-check exits cleanly.
+        while (!inbox.offerLast(error)) {
+            if (inbox.pollFirst() != null) {
+                dropped.incrementAndGet();
+            }
+            if (closed) {
+                return false;
+            }
         }
         // Common case after the first offer: thread already running, hot
         // path is one volatile read. Lazy start happens once per dispatcher

core/src/main/java/io/questdb/client/std/Files.java

@@ -258,12 +258,15 @@ public static void freeNativePath(long pathPtr) {
      */
     public static int rename(String oldPath, String newPath) {
         long o = pathPtr(oldPath);
-        long n = pathPtr(newPath);
         try {
-            return rename0(o, n);
+            long n = pathPtr(newPath);
+            try {
+                return rename0(o, n);
+            } finally {
+                freePathPtr(n);
+            }
         } finally {
             freePathPtr(o);
-            freePathPtr(n);
         }
     }