feat(qwp): make the poison-frame threshold configurable (max_frame_rejections)

Turns CursorWebSocketSendLoop.MAX_HEAD_FRAME_REJECTIONS (renamed
DEFAULT_MAX_HEAD_FRAME_REJECTIONS = 4) into per-sender configuration:
consecutive server-active rejections (retriable NACK or non-orderly
close) of the same head-of-line frame, with no ack progress in between,
before the sender declares the frame poisoned and latches a typed
terminal instead of reconnect-replaying forever.

Configurable from every surface:
- connect string: max_frame_rejections=N (>= 1), both the hand parser
  and the ConfigView path; registered as an INGRESS key in ConfigSchema,
  so the QwpQueryClient accepts it as a syntactic no-op (one vocabulary,
  side-owned application -- pinned by QwpConfigKeysTest for both
  clients).
- programmatic: LineSenderBuilder.maxFrameRejections(int), WebSocket
  only, validated >= 1.
- QuestDB facade: the config string flows through the pool to every
  built Sender; an out-of-range value is rejected at build() even with
  min=0 pools (QuestDBBuilderTest).

The threshold is forwarded to the cursor send loop AND to every
BackgroundDrainer (the drainer replays the owner's SF data, so it must
honor the same threshold). All existing constructors are preserved as
delegating overloads with the default.

Tests: WsSenderConfigHonoredTest drift guard (key applied, not merely
accepted); ServerErrorAckTerminalTest e2e proves max_frame_rejections=2
escalates after exactly two deliveries of the same frame instead of the
default four.

Author: bluestreak01

.gitignore

@@ -35,3 +35,4 @@ core/CMakeCache.txt
 
 # pi subagents runtime artifacts
 .pi-subagents/
+core/src/main/resources/io/questdb/client/bin/

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

@@ -1049,6 +1049,7 @@ final class LineSenderBuilder {
         // Bounded inbox capacity for the async error dispatcher.
         // PARAMETER_NOT_SET_EXPLICITLY → spec default (256).
         private int errorInboxCapacity = PARAMETER_NOT_SET_EXPLICITLY;
+        private int maxFrameRejections = PARAMETER_NOT_SET_EXPLICITLY;
         private String httpPath;
         private String httpSettingsPath;
         private int httpTimeout = PARAMETER_NOT_SET_EXPLICITLY;
@@ -1497,6 +1498,9 @@ public Sender build() {
                         durableAckKeepaliveIntervalMillis == DURABLE_ACK_KEEPALIVE_NOT_SET
                                 ? CursorWebSocketSendLoop.DEFAULT_DURABLE_ACK_KEEPALIVE_INTERVAL_MILLIS
                                 : durableAckKeepaliveIntervalMillis;
+                int actualMaxFrameRejections = maxFrameRejections != PARAMETER_NOT_SET_EXPLICITLY
+                        ? maxFrameRejections
+                        : CursorWebSocketSendLoop.DEFAULT_MAX_HEAD_FRAME_REJECTIONS;
 
                 // sfDir is the parent (group root); the actual slot lives
                 // under sfDir/senderId. This is what the engine sees — the
@@ -1568,7 +1572,8 @@ public Sender build() {
                             authTimeoutMillis,
                             connectTimeoutMillis == PARAMETER_NOT_SET_EXPLICITLY ? 0 : connectTimeoutMillis,
                             connectionListener,
-                            actualConnectionListenerInboxCapacity
+                            actualConnectionListenerInboxCapacity,
+                            actualMaxFrameRejections
                     );
                 } catch (Throwable t) {
                     // connect() failed before ownership of cursorEngine
@@ -2417,6 +2422,27 @@ public LineSenderBuilder reconnectInitialBackoffMillis(long millis) {
             return this;
         }
 
+        /**
+         * Poison-frame detector threshold: consecutive server rejections
+         * (retriable NACK, or non-orderly close after a send) of the SAME
+         * head-of-line frame, with no ack progress in between, before the
+         * sender declares the frame poisoned and latches a typed terminal
+         * instead of reconnect-replaying forever. Retriable rejections below
+         * the threshold recycle the connection and replay from the
+         * store-and-forward log — no data is dropped either way.
+         * Default {@code 4}. WebSocket only.
+         */
+        public LineSenderBuilder maxFrameRejections(int rejections) {
+            if (protocol != PARAMETER_NOT_SET_EXPLICITLY && protocol != PROTOCOL_WEBSOCKET) {
+                throw new LineSenderException("max_frame_rejections is only supported for WebSocket transport");
+            }
+            if (rejections < 1) {
+                throw new LineSenderException("max_frame_rejections must be >= 1: ").put(rejections);
+            }
+            this.maxFrameRejections = rejections;
+            return this;
+        }
+
         /**
          * Max reconnect backoff in millis. Caps the exponential growth so
          * a long outage doesn't end up sleeping minutes between attempts.
@@ -3342,6 +3368,12 @@ private LineSenderBuilder fromConfig(CharSequence configurationString) {
                     }
                     pos = getValue(configurationString, pos, sink, "reconnect_initial_backoff_millis");
                     reconnectInitialBackoffMillis(parseLongValue(sink, "reconnect_initial_backoff_millis"));
+                } else if (Chars.equals("max_frame_rejections", sink)) {
+                    if (protocol != PROTOCOL_WEBSOCKET) {
+                        throw new LineSenderException("max_frame_rejections is only supported for WebSocket transport");
+                    }
+                    pos = getValue(configurationString, pos, sink, "max_frame_rejections");
+                    maxFrameRejections(parseIntValue(sink, "max_frame_rejections"));
                 } else if (Chars.equals("initial_connect_retry", sink)) {
                     if (protocol != PROTOCOL_WEBSOCKET) {
                         throw new LineSenderException("initial_connect_retry is only supported for WebSocket transport");
@@ -3607,6 +3639,9 @@ private LineSenderBuilder fromConfigWebSocket(CharSequence configurationString)
                 if (view.has("reconnect_max_backoff_millis")) {
                     reconnectMaxBackoffMillis(wsLong(view, v, "reconnect_max_backoff_millis"));
                 }
+                if (view.has("max_frame_rejections")) {
+                    maxFrameRejections(wsInt(view, v, "max_frame_rejections"));
+                }
                 if (view.has("sf_append_deadline_millis")) {
                     sfAppendDeadlineMillis(wsLong(view, v, "sf_append_deadline_millis"));
                 }
@@ -3781,6 +3816,7 @@ public java.util.Map<String, Object> wsConfigSnapshotForTest() {
             m.put("reconnect_max_backoff_millis", reconnectMaxBackoffMillis);
             m.put("drain_orphans", drainOrphans);
             m.put("max_background_drainers", maxBackgroundDrainers);
+            m.put("max_frame_rejections", maxFrameRejections);
             m.put("error_inbox_capacity", errorInboxCapacity);
             m.put("connection_listener_inbox_capacity", connectionListenerInboxCapacity);
             m.put("token", httpToken);

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

@@ -205,7 +205,7 @@ public enum Category {
         /**
          * A frame the server (or an intermediary) deterministically rejects: the
          * poison-frame detector observed the same head-of-line frame fail
-         * {@link io.questdb.client.cutlass.qwp.client.sf.cursor.CursorWebSocketSendLoop#MAX_HEAD_FRAME_REJECTIONS}
+         * {@link io.questdb.client.cutlass.qwp.client.sf.cursor.CursorWebSocketSendLoop#DEFAULT_MAX_HEAD_FRAME_REJECTIONS}
          * consecutive times with no ack progress — replaying it cannot succeed.
          */
         PROTOCOL_VIOLATION,

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

@@ -281,6 +281,9 @@ public class QwpWebSocketSender implements Sender {
     // Async-delivery sink for ack-watermark advances. Default no-op; a
     // setProgressHandler call before connect() swaps in a real handler.
     private SenderProgressHandler progressHandler = DefaultSenderProgressHandler.INSTANCE;
+    // Poison-frame detector threshold forwarded to the cursor send loop and to
+    // every background drainer (connect-string key max_frame_rejections).
+    private int maxFrameRejections = CursorWebSocketSendLoop.DEFAULT_MAX_HEAD_FRAME_REJECTIONS;
     private long reconnectInitialBackoffMillis =
             CursorWebSocketSendLoop.DEFAULT_RECONNECT_INITIAL_BACKOFF_MILLIS;
     private long reconnectMaxBackoffMillis =
@@ -620,7 +623,8 @@ public static QwpWebSocketSender connect(
 
     /**
      * Multi-endpoint variant that also accepts the async connection-event
-     * listener and its dispatcher inbox capacity.
+     * listener and its dispatcher inbox capacity. Uses the default
+     * poison-frame detector threshold.
      */
     public static QwpWebSocketSender connect(
             List<Endpoint> endpoints,
@@ -643,6 +647,45 @@ public static QwpWebSocketSender connect(
             int connectTimeoutMs,
             SenderConnectionListener connectionListener,
             int connectionListenerInboxCapacity
+    ) {
+        return connect(endpoints, tlsConfig, autoFlushRows, autoFlushBytes,
+                autoFlushIntervalNanos, authorizationHeader, requestDurableAck,
+                cursorEngine, closeFlushTimeoutMillis, reconnectMaxDurationMillis,
+                reconnectInitialBackoffMillis, reconnectMaxBackoffMillis,
+                initialConnectMode, errorHandler, errorInboxCapacity,
+                durableAckKeepaliveIntervalMillis, authTimeoutMs, connectTimeoutMs,
+                connectionListener, connectionListenerInboxCapacity,
+                CursorWebSocketSendLoop.DEFAULT_MAX_HEAD_FRAME_REJECTIONS);
+    }
+
+    /**
+     * Master connect overload — also accepts the poison-frame detector
+     * threshold ({@code max_frame_rejections}): consecutive server-active
+     * rejections of the same head-of-line frame, with no ack progress in
+     * between, before the loop escalates to a typed terminal.
+     */
+    public static QwpWebSocketSender connect(
+            List<Endpoint> endpoints,
+            ClientTlsConfiguration tlsConfig,
+            int autoFlushRows,
+            int autoFlushBytes,
+            long autoFlushIntervalNanos,
+            String authorizationHeader,
+            boolean requestDurableAck,
+            CursorSendEngine cursorEngine,
+            long closeFlushTimeoutMillis,
+            long reconnectMaxDurationMillis,
+            long reconnectInitialBackoffMillis,
+            long reconnectMaxBackoffMillis,
+            Sender.InitialConnectMode initialConnectMode,
+            SenderErrorHandler errorHandler,
+            int errorInboxCapacity,
+            long durableAckKeepaliveIntervalMillis,
+            long authTimeoutMs,
+            int connectTimeoutMs,
+            SenderConnectionListener connectionListener,
+            int connectionListenerInboxCapacity,
+            int maxFrameRejections
     ) {
         QwpWebSocketSender sender = new QwpWebSocketSender(
                 endpoints, tlsConfig,
@@ -658,6 +701,7 @@ public static QwpWebSocketSender connect(
             sender.reconnectInitialBackoffMillis = reconnectInitialBackoffMillis;
             sender.reconnectMaxBackoffMillis = reconnectMaxBackoffMillis;
             sender.durableAckKeepaliveIntervalMillis = durableAckKeepaliveIntervalMillis;
+            sender.maxFrameRejections = maxFrameRejections;
             sender.initialConnectMode = initialConnectMode == null
                     ? Sender.InitialConnectMode.OFF
                     : initialConnectMode;
@@ -2308,7 +2352,8 @@ public synchronized void startOrphanDrainers(
                             reconnectInitialBackoffMillis,
                             reconnectMaxBackoffMillis,
                             requestDurableAck,
-                            durableAckKeepaliveIntervalMillis);
+                            durableAckKeepaliveIntervalMillis,
+                            maxFrameRejections);
             ref[0] = drainer;
             drainerPool.submit(drainer);
         }
@@ -3158,7 +3203,8 @@ private void ensureConnected() {
                     reconnectInitialBackoffMillis,
                     reconnectMaxBackoffMillis,
                     requestDurableAck,
-                    durableAckKeepaliveIntervalMillis);
+                    durableAckKeepaliveIntervalMillis,
+                    maxFrameRejections);
             // Plug the async-delivery sink before start() so the I/O thread
             // never observes a null dispatcher between recordFatal and
             // notification — the test for null in dispatchError handles

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

@@ -126,6 +126,9 @@ public final class BackgroundDrainer implements Runnable {
      */
     private volatile Thread runnerThread;
     private volatile boolean stopRequested;
+    // Poison-frame detector threshold forwarded to every drain loop this
+    // drainer creates; mirrors the owner sender's max_frame_rejections config.
+    private final int maxHeadFrameRejections;
 
     public BackgroundDrainer(
             String slotPath,
@@ -137,6 +140,31 @@ public BackgroundDrainer(
             long reconnectMaxBackoffMillis,
             boolean requestDurableAck,
             long durableAckKeepaliveIntervalMillis
+    ) {
+        this(slotPath, segmentSizeBytes, sfMaxTotalBytes, clientFactory,
+                reconnectMaxDurationMillis, reconnectInitialBackoffMillis,
+                reconnectMaxBackoffMillis, requestDurableAck,
+                durableAckKeepaliveIntervalMillis,
+                CursorWebSocketSendLoop.DEFAULT_MAX_HEAD_FRAME_REJECTIONS);
+    }
+
+    /**
+     * Master constructor — also accepts the poison-frame detector threshold
+     * ({@code max_frame_rejections}) forwarded to the drain loop's
+     * {@link CursorWebSocketSendLoop}: the drainer replays the owner sender's
+     * SF data, so it must honor the same configured threshold.
+     */
+    public BackgroundDrainer(
+            String slotPath,
+            long segmentSizeBytes,
+            long sfMaxTotalBytes,
+            CursorWebSocketSendLoop.ReconnectFactory clientFactory,
+            long reconnectMaxDurationMillis,
+            long reconnectInitialBackoffMillis,
+            long reconnectMaxBackoffMillis,
+            boolean requestDurableAck,
+            long durableAckKeepaliveIntervalMillis,
+            int maxHeadFrameRejections
     ) {
         this.slotPath = slotPath;
         this.segmentSizeBytes = segmentSizeBytes;
@@ -147,6 +175,7 @@ public BackgroundDrainer(
         this.reconnectMaxBackoffMillis = reconnectMaxBackoffMillis;
         this.requestDurableAck = requestDurableAck;
         this.durableAckKeepaliveIntervalMillis = durableAckKeepaliveIntervalMillis;
+        this.maxHeadFrameRejections = maxHeadFrameRejections;
     }
 
     /**
@@ -506,7 +535,8 @@ public void run() {
                         reconnectInitialBackoffMillis,
                         reconnectMaxBackoffMillis,
                         requestDurableAck,
-                        durableAckKeepaliveIntervalMillis);
+                        durableAckKeepaliveIntervalMillis,
+                        maxHeadFrameRejections);
                 loop.start();
 
                 while (!stopRequested) {