fix(qwp): pace transient close recycles + minimum poison-escalation dwell

The poison-frame detector could turn a TRANSIENT, post-connect outage into a
producer-fatal PROTOCOL_VIOLATION -- the exact false-positive its design
contract forbids ("a genuine outage fails at connect, not deterministically on
one FSN"). A middlebox/LB that completes the WS upgrade, accepts the head frame,
then non-orderly-closes (1006/1011) while its backend is briefly down succeeds
at CONNECT every cycle, so connectLoop's failed-connect backoff never engages.
Two gaps let that latch a terminal:

1. Below the strike threshold, onClose recycled through the UNPACED fail() --
   unlike the NACK path, which uses failPaced(). Strikes then accrued at
   connect+send+close RTT rate (well under a second) with no widening gaps.
   Fix: route the below-threshold non-orderly-close strike through the same
   strike-escalated failPaced the NACK path uses; genuine transport closes
   (orderly, or before any send) still reconnect immediately via fail().

2. Escalation fired on strike count alone, so even with pacing a brief outage
   (longer than the ~sub-second strike window) still escalated. Fix: gate
   escalation on a minimum wall-clock dwell -- poisonStrikes >= threshold AND
   the suspect has stayed poisoned for at least poison_min_escalation_window_
   millis (default 5000ms). This decouples "how many strikes prove determinism"
   from "how long a transient is allowed to look poisoned"; an OK at/beyond the
   suspect during the window resets the detector. 0 = legacy immediate
   escalation at the threshold.

The dwell is first-class config, mirroring max_frame_rejections end-to-end:
connect-string key poison_min_escalation_window_millis (ConfigSchema INGRESS),
LineSenderBuilder.poisonMinEscalationWindowMillis(long), forwarded to the cursor
send loop and every BackgroundDrainer. The raw CursorWebSocketSendLoop
constructor overloads default the dwell to 0 (legacy) so existing tests keep
exact escalate-at-N semantics; the 5s user default is applied at the config
layer, same split max_frame_rejections uses.

Tests (CursorWebSocketSendLoopPoisonFrameTest):
- testNonOrderlyCloseRecycleIsPacedAgainstAcceptingMiddlebox: real I/O thread;
  unpaced code records ~200k recycles/1.2s, paced <=10. Proven red->green.
- testPoisonDwellHoldsEscalationUntilWallClockWindowElapses: strikes past the
  threshold must not escalate until the window elapses. Proven red->green.
- ServerErrorAckTerminalTest escalation tests set
  poison_min_escalation_window_millis=0 so exact per-strike counts stay
  assertable; WsSenderConfigHonoredTest pins the new key end-to-end.

Author: bluestreak01

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

@@ -1050,6 +1050,7 @@ final class LineSenderBuilder {
         // PARAMETER_NOT_SET_EXPLICITLY → spec default (256).
         private int errorInboxCapacity = PARAMETER_NOT_SET_EXPLICITLY;
         private int maxFrameRejections = PARAMETER_NOT_SET_EXPLICITLY;
+        private long poisonMinEscalationWindowMillis = PARAMETER_NOT_SET_EXPLICITLY;
         private String httpPath;
         private String httpSettingsPath;
         private int httpTimeout = PARAMETER_NOT_SET_EXPLICITLY;
@@ -1501,6 +1502,9 @@ public Sender build() {
                 int actualMaxFrameRejections = maxFrameRejections != PARAMETER_NOT_SET_EXPLICITLY
                         ? maxFrameRejections
                         : CursorWebSocketSendLoop.DEFAULT_MAX_HEAD_FRAME_REJECTIONS;
+                long actualPoisonMinEscalationWindowMillis = poisonMinEscalationWindowMillis != PARAMETER_NOT_SET_EXPLICITLY
+                        ? poisonMinEscalationWindowMillis
+                        : CursorWebSocketSendLoop.DEFAULT_POISON_MIN_ESCALATION_WINDOW_MILLIS;
 
                 // sfDir is the parent (group root); the actual slot lives
                 // under sfDir/senderId. This is what the engine sees — the
@@ -1573,7 +1577,8 @@ public Sender build() {
                             connectTimeoutMillis == PARAMETER_NOT_SET_EXPLICITLY ? 0 : connectTimeoutMillis,
                             connectionListener,
                             actualConnectionListenerInboxCapacity,
-                            actualMaxFrameRejections
+                            actualMaxFrameRejections,
+                            actualPoisonMinEscalationWindowMillis
                     );
                 } catch (Throwable t) {
                     // connect() failed before ownership of cursorEngine
@@ -2443,6 +2448,30 @@ public LineSenderBuilder maxFrameRejections(int rejections) {
             return this;
         }
 
+        /**
+         * Minimum wall-clock dwell (millis) a suspected frame must stay
+         * poisoned before the poison detector escalates to a typed terminal,
+         * even once {@link #maxFrameRejections(int)} strikes have accrued.
+         * With paced recycles, strikes can accrue in well under a second (e.g.
+         * a middlebox/LB that accepts the connection then non-orderly-closes
+         * each cycle while its backend is briefly down), so a strike count
+         * alone can turn a transient outage into a producer-fatal terminal.
+         * This window guarantees a brief outage a chance to clear (an OK
+         * at/beyond the suspect resets the detector) first. {@code 0} disables
+         * the dwell (legacy immediate escalation at the strike threshold).
+         * Default {@code 5_000} (5 s). WebSocket only.
+         */
+        public LineSenderBuilder poisonMinEscalationWindowMillis(long millis) {
+            if (protocol != PARAMETER_NOT_SET_EXPLICITLY && protocol != PROTOCOL_WEBSOCKET) {
+                throw new LineSenderException("poison_min_escalation_window_millis is only supported for WebSocket transport");
+            }
+            if (millis < 0) {
+                throw new LineSenderException("poison_min_escalation_window_millis must be >= 0: ").put(millis);
+            }
+            this.poisonMinEscalationWindowMillis = millis;
+            return this;
+        }
+
         /**
          * Max reconnect backoff in millis. Caps the exponential growth so
          * a long outage doesn't end up sleeping minutes between attempts.
@@ -3378,6 +3407,12 @@ private LineSenderBuilder fromConfig(CharSequence configurationString) {
                     }
                     pos = getValue(configurationString, pos, sink, "max_frame_rejections");
                     maxFrameRejections(parseIntValue(sink, "max_frame_rejections"));
+                } else if (Chars.equals("poison_min_escalation_window_millis", sink)) {
+                    if (protocol != PROTOCOL_WEBSOCKET) {
+                        throw new LineSenderException("poison_min_escalation_window_millis is only supported for WebSocket transport");
+                    }
+                    pos = getValue(configurationString, pos, sink, "poison_min_escalation_window_millis");
+                    poisonMinEscalationWindowMillis(parseLongValue(sink, "poison_min_escalation_window_millis"));
                 } else if (Chars.equals("initial_connect_retry", sink)) {
                     if (protocol != PROTOCOL_WEBSOCKET) {
                         throw new LineSenderException("initial_connect_retry is only supported for WebSocket transport");
@@ -3646,6 +3681,9 @@ private LineSenderBuilder fromConfigWebSocket(CharSequence configurationString)
                 if (view.has("max_frame_rejections")) {
                     maxFrameRejections(wsInt(view, v, "max_frame_rejections"));
                 }
+                if (view.has("poison_min_escalation_window_millis")) {
+                    poisonMinEscalationWindowMillis(wsLong(view, v, "poison_min_escalation_window_millis"));
+                }
                 if (view.has("sf_append_deadline_millis")) {
                     sfAppendDeadlineMillis(wsLong(view, v, "sf_append_deadline_millis"));
                 }
@@ -3821,6 +3859,7 @@ public java.util.Map<String, Object> wsConfigSnapshotForTest() {
             m.put("drain_orphans", drainOrphans);
             m.put("max_background_drainers", maxBackgroundDrainers);
             m.put("max_frame_rejections", maxFrameRejections);
+            m.put("poison_min_escalation_window_millis", poisonMinEscalationWindowMillis);
             m.put("error_inbox_capacity", errorInboxCapacity);
             m.put("connection_listener_inbox_capacity", connectionListenerInboxCapacity);
             m.put("token", httpToken);

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

@@ -284,6 +284,10 @@ public class QwpWebSocketSender implements Sender {
     // 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;
+    // Minimum wall-clock dwell before poison escalation, forwarded alongside
+    // maxFrameRejections (connect-string key poison_min_escalation_window_millis).
+    private long poisonMinEscalationWindowMillis =
+            CursorWebSocketSendLoop.DEFAULT_POISON_MIN_ESCALATION_WINDOW_MILLIS;
     private long reconnectInitialBackoffMillis =
             CursorWebSocketSendLoop.DEFAULT_RECONNECT_INITIAL_BACKOFF_MILLIS;
     private long reconnectMaxBackoffMillis =
@@ -655,7 +659,8 @@ public static QwpWebSocketSender connect(
                 initialConnectMode, errorHandler, errorInboxCapacity,
                 durableAckKeepaliveIntervalMillis, authTimeoutMs, connectTimeoutMs,
                 connectionListener, connectionListenerInboxCapacity,
-                CursorWebSocketSendLoop.DEFAULT_MAX_HEAD_FRAME_REJECTIONS);
+                CursorWebSocketSendLoop.DEFAULT_MAX_HEAD_FRAME_REJECTIONS,
+                CursorWebSocketSendLoop.DEFAULT_POISON_MIN_ESCALATION_WINDOW_MILLIS);
     }
 
     /**
@@ -685,7 +690,8 @@ public static QwpWebSocketSender connect(
             int connectTimeoutMs,
             SenderConnectionListener connectionListener,
             int connectionListenerInboxCapacity,
-            int maxFrameRejections
+            int maxFrameRejections,
+            long poisonMinEscalationWindowMillis
     ) {
         QwpWebSocketSender sender = new QwpWebSocketSender(
                 endpoints, tlsConfig,
@@ -702,6 +708,7 @@ public static QwpWebSocketSender connect(
             sender.reconnectMaxBackoffMillis = reconnectMaxBackoffMillis;
             sender.durableAckKeepaliveIntervalMillis = durableAckKeepaliveIntervalMillis;
             sender.maxFrameRejections = maxFrameRejections;
+            sender.poisonMinEscalationWindowMillis = poisonMinEscalationWindowMillis;
             sender.initialConnectMode = initialConnectMode == null
                     ? Sender.InitialConnectMode.OFF
                     : initialConnectMode;
@@ -2353,7 +2360,8 @@ public synchronized void startOrphanDrainers(
                             reconnectMaxBackoffMillis,
                             requestDurableAck,
                             durableAckKeepaliveIntervalMillis,
-                            maxFrameRejections);
+                            maxFrameRejections,
+                            poisonMinEscalationWindowMillis);
             ref[0] = drainer;
             drainerPool.submit(drainer);
         }
@@ -3211,7 +3219,8 @@ private void ensureConnected() {
                     reconnectMaxBackoffMillis,
                     requestDurableAck,
                     durableAckKeepaliveIntervalMillis,
-                    maxFrameRejections);
+                    maxFrameRejections,
+                    poisonMinEscalationWindowMillis);
             // 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

@@ -129,6 +129,9 @@ public final class BackgroundDrainer implements Runnable {
     // 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;
+    // Minimum wall-clock dwell before poison escalation, forwarded to every
+    // drain loop; mirrors the owner sender's poison_min_escalation_window_millis.
+    private final long poisonMinEscalationWindowMillis;
 
     public BackgroundDrainer(
             String slotPath,
@@ -145,7 +148,8 @@ public BackgroundDrainer(
                 reconnectMaxDurationMillis, reconnectInitialBackoffMillis,
                 reconnectMaxBackoffMillis, requestDurableAck,
                 durableAckKeepaliveIntervalMillis,
-                CursorWebSocketSendLoop.DEFAULT_MAX_HEAD_FRAME_REJECTIONS);
+                CursorWebSocketSendLoop.DEFAULT_MAX_HEAD_FRAME_REJECTIONS,
+                CursorWebSocketSendLoop.DEFAULT_POISON_MIN_ESCALATION_WINDOW_MILLIS);
     }
 
     /**
@@ -164,7 +168,8 @@ public BackgroundDrainer(
             long reconnectMaxBackoffMillis,
             boolean requestDurableAck,
             long durableAckKeepaliveIntervalMillis,
-            int maxHeadFrameRejections
+            int maxHeadFrameRejections,
+            long poisonMinEscalationWindowMillis
     ) {
         this.slotPath = slotPath;
         this.segmentSizeBytes = segmentSizeBytes;
@@ -176,6 +181,7 @@ public BackgroundDrainer(
         this.requestDurableAck = requestDurableAck;
         this.durableAckKeepaliveIntervalMillis = durableAckKeepaliveIntervalMillis;
         this.maxHeadFrameRejections = maxHeadFrameRejections;
+        this.poisonMinEscalationWindowMillis = poisonMinEscalationWindowMillis;
     }
 
     /**
@@ -187,7 +193,8 @@ public BackgroundDrainer(
      */
     @TestOnly
     public BackgroundDrainer() {
-        this(null, 0L, 0L, null, 0L, 0L, 0L, false, 0L);
+        this(null, 0L, 0L, null, 0L, 0L, 0L, false, 0L,
+                CursorWebSocketSendLoop.DEFAULT_MAX_HEAD_FRAME_REJECTIONS, 0L);
     }
 
     /**
@@ -567,7 +574,8 @@ public void run() {
                         reconnectMaxBackoffMillis,
                         requestDurableAck,
                         durableAckKeepaliveIntervalMillis,
-                        maxHeadFrameRejections);
+                        maxHeadFrameRejections,
+                        poisonMinEscalationWindowMillis);
                 loop.start();
 
                 while (!stopRequestedOrInterrupted()) {