fix(qwp): cursor I/O loop dies unlatched when the reconnect machinery throws

A RuntimeException escaping fail()/connectLoop pierced ioLoop's catch-all
and killed the I/O thread with running still true and no terminal latched:
checkError() stayed clean and the producer kept buffering into SF against
a dead loop -- the exact silent stall Invariant B exists to prevent. The
concrete trigger was the retry-loop jitter: ThreadLocalRandom.nextLong
throws IAE on a bound <= 0, and the loop constructors (unlike the builder)
do not validate the backoff params, so direct construction with a
non-positive backoff turned the first backoff into an unlatched thread
death.

Fixed twice over:
- clamp both jitter bounds with Math.max(1L, ...), mirroring
  BackgroundDrainer's sweep-loop guard;
- guard the fail(t) call in ioLoop's catch: any secondary failure now
  latches a terminal (latch-and-die, like the Error arm) with the original
  failure attached as suppressed, instead of escaping. recordFatal is
  first-writer-wins, so a terminal already latched inside connectLoop is
  preserved.

CursorWebSocketSendLoopZeroBackoffTest is red pre-fix on both counts:
the I/O thread died after 2 attempts (IAE: bound must be positive) with
no terminal latched, and connectWithRetry surfaced the raw IAE instead
of budget exhaustion.

Author: bluestreak01

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

@@ -505,7 +505,10 @@ public static WebSocketClient connectWithRetry(
                     lastLogNanos = now;
                 }
             }
-            long jitter = ThreadLocalRandom.current().nextLong(backoffMillis);
+            // Math.max guards nextLong's bound>0 contract (IAE on <= 0)
+            // against non-builder callers -- builder-validated config keeps
+            // this > 0. Mirrors BackgroundDrainer's sweep-loop jitter guard.
+            long jitter = ThreadLocalRandom.current().nextLong(Math.max(1L, backoffMillis));
             long sleepMillis = backoffMillis + jitter;
             long remainingMillis = (deadlineNanos - System.nanoTime()) / 1_000_000L;
             if (remainingMillis <= 0) {
@@ -1171,7 +1174,13 @@ private void connectLoop(Throwable initial, String phase) {
                 }
             }
             if (running) {
-                long jitter = ThreadLocalRandom.current().nextLong(backoffMillis);
+                // Math.max guards nextLong's bound>0 contract (IAE on <= 0):
+                // an IAE here would climb through fail() into ioLoop's
+                // secondary-failure guard and latch a terminal -- correct but
+                // needlessly fatal for a jitter computation. Builder-validated
+                // config keeps this > 0; the guard covers direct constructor
+                // use. Mirrors BackgroundDrainer's sweep-loop jitter guard.
+                long jitter = ThreadLocalRandom.current().nextLong(Math.max(1L, backoffMillis));
                 long sleepMillis = backoffMillis + jitter;
                 LockSupport.parkNanos(sleepMillis * 1_000_000L);
                 backoffMillis = Math.min(backoffMillis * 2, reconnectMaxBackoffMillis);
@@ -1382,7 +1391,28 @@ private void ioLoop() {
                 recordFatal(t);
                 throw (Error) t;
             }
-            fail(t);
+            try {
+                fail(t);
+            } catch (Throwable secondary) {
+                // fail()/connectLoop itself threw. Without this guard the
+                // secondary failure escapes ioLoop's catch and kills the I/O
+                // thread with `running` still true and no terminal latched:
+                // checkError() stays clean and the producer keeps buffering
+                // into SF against a dead loop -- the exact silent stall
+                // Invariant B exists to prevent. Latch-and-die like the Error
+                // arm above instead; the finally still counts down the
+                // shutdown latch, so close() cannot hang. recordFatal is
+                // first-writer-wins, so a terminal already latched inside
+                // connectLoop (e.g. a throw from dispatchError after
+                // recordFatal) is preserved.
+                if (secondary != t) {
+                    secondary.addSuppressed(t);
+                }
+                recordFatal(secondary);
+                if (secondary instanceof Error) {
+                    throw (Error) secondary;
+                }
+            }
         } finally {
             // Last act of the I/O thread: dispose of whatever client it
             // holds. This is the airtight half of the close()-vs-reconnect

core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoopZeroBackoffTest.java

@@ -0,0 +1,138 @@
+/*+*****************************************************************************
+ *     ___                  _   ____  ____
+ *    / _ \ _   _  ___  ___| |_|  _ \| __ )
+ *   | | | | | | |/ _ \/ __| __| | | |  _ \
+ *   | |_| | |_| |  __/\__ \ |_| |_| | |_) |
+ *    \__\_\\__,_|\___||___/\__|____/|____/
+ *
+ *  Copyright (c) 2014-2019 Appsicle
+ *  Copyright (c) 2019-2026 QuestDB
+ *
+ *  Licensed under the Apache License, Version 2.0 (the "License");
+ *  you may not use this file except in compliance with the License.
+ *  You may obtain a copy of the License at
+ *
+ *  http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ *
+ ******************************************************************************/
+
+package io.questdb.client.test.cutlass.qwp.client.sf.cursor;
+
+import io.questdb.client.cutlass.line.LineSenderException;
+import io.questdb.client.cutlass.qwp.client.sf.cursor.CursorSendEngine;
+import io.questdb.client.cutlass.qwp.client.sf.cursor.CursorWebSocketSendLoop;
+import org.junit.Assert;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+import java.io.IOException;
+
+/**
+ * Regression guard for the I/O-thread silent-death hazard in
+ * {@link CursorWebSocketSendLoop}: a {@code RuntimeException} escaping
+ * {@code fail()}/{@code connectLoop} pierced {@code ioLoop}'s catch-all and
+ * killed the I/O thread with {@code running} still {@code true} and no
+ * terminal latched — the producer kept buffering into SF against a dead loop,
+ * the exact silent stall Invariant B exists to prevent.
+ * <p>
+ * The concrete trigger was the retry-loop jitter:
+ * {@code ThreadLocalRandom.nextLong(backoffMillis)} throws
+ * {@code IllegalArgumentException} when the bound is {@code <= 0}. The
+ * builder validates {@code reconnect_initial_backoff_millis > 0}, but the
+ * loop constructors are public and do not — direct construction with a
+ * non-positive backoff (as this test does) turned the very first backoff
+ * into an unlatched thread death. Fixed twice over: the jitter bounds are
+ * clamped with {@code Math.max(1L, …)} (mirroring {@code BackgroundDrainer}'s
+ * sweep loop), and {@code ioLoop}'s catch guards the {@code fail(t)} call so
+ * any residual secondary failure latches a terminal (latch-and-die) instead
+ * of escaping.
+ */
+public class CursorWebSocketSendLoopZeroBackoffTest {
+
+    @Rule
+    public final TemporaryFolder sfDir = TemporaryFolder.builder().assureDeletion().build();
+
+    /**
+     * Pins the static bounded-connect variant ({@code connectWithRetry},
+     * user-thread blocking initial connect): a non-positive initial backoff
+     * must exhaust the budget and surface the real (transport) failure as a
+     * {@link LineSenderException} — not blow up in the jitter computation
+     * with an {@code IllegalArgumentException}.
+     */
+    @Test(timeout = 30_000)
+    public void connectWithRetryToleratesNonPositiveBackoff() {
+        try {
+            CursorWebSocketSendLoop.connectWithRetry(
+                    () -> {
+                        throw new IOException("connection refused (test)");
+                    },
+                    50L,  // budget: fail fast
+                    0L,   // non-positive initial backoff: pre-fix IAE in nextLong
+                    1L,
+                    "test-connect");
+            Assert.fail("expected LineSenderException on budget exhaustion");
+        } catch (LineSenderException expected) {
+            Assert.assertTrue(
+                    "budget exhaustion should carry the attempt summary, got: "
+                            + expected.getMessage(),
+                    expected.getMessage().contains("failed after"));
+        }
+    }
+
+    /**
+     * Pre-fix: the I/O thread walked the async-initial-connect retry loop,
+     * hit {@code nextLong(0)} on the first backoff, and the IAE — re-thrown
+     * once more from the {@code fail(t)} re-entry — escaped {@code ioLoop}'s
+     * catch. The thread died after at most two recorded attempts with
+     * {@code isRunning() == true} and {@code getTerminalError() == null}:
+     * invisible to the producer. Post-fix the clamped jitter keeps the loop
+     * retrying indefinitely (Invariant B), so attempts keep climbing and no
+     * terminal is latched.
+     */
+    @Test(timeout = 30_000)
+    public void zeroBackoffConnectFailuresKeepRetryingInsteadOfKillingIoThread() throws Exception {
+        try (CursorSendEngine engine = new CursorSendEngine(
+                sfDir.getRoot().getAbsolutePath(), 16_384)) {
+            CursorWebSocketSendLoop loop = new CursorWebSocketSendLoop(
+                    null, engine, 0, 1_000_000L,
+                    () -> {
+                        // Plain connect failure: retried forever under
+                        // Invariant B — never terminal, never latched.
+                        throw new IOException("connection refused (test)");
+                    },
+                    0,
+                    0,   // non-positive initial backoff: pre-fix IAE in nextLong
+                    1);
+            try {
+                loop.start();
+                // Post-fix the zero-backoff loop records attempts at a very
+                // high rate; pre-fix the count freezes at <= 2 when the IAE
+                // kills the thread, and this poll times out.
+                long deadlineNanos = System.nanoTime() + 20_000_000_000L;
+                while (loop.getTotalReconnectAttempts() < 5
+                        && System.nanoTime() < deadlineNanos) {
+                    Thread.yield();
+                }
+                Assert.assertTrue(
+                        "I/O thread stopped retrying after "
+                                + loop.getTotalReconnectAttempts()
+                                + " attempts -- it died without latching a terminal "
+                                + "(the Invariant B silent stall)",
+                        loop.getTotalReconnectAttempts() >= 5);
+                Assert.assertTrue("loop must still be running", loop.isRunning());
+                Assert.assertNull(
+                        "plain connect failures must not latch a terminal (Invariant B)",
+                        loop.getTerminalError());
+            } finally {
+                loop.close();
+            }
+        }
+    }
+}