test(qwp): make Invariant-B tests detect a silent retry-loop give-up

The "retries forever" tests asserted only no-terminal-in-window, but a
regressed I/O thread that gives up WITHOUT latching a terminal also
delivers nothing: flush() publishes to the cursor engine on the user
thread regardless of wire liveness, so both tests passed green with a
dead retry loop.

Back the no-terminal assertions with two discriminators:

- LIVENESS: snapshot getTotalReconnectAttempts() only after the
  (ignored) reconnect budget has elapsed and require it to advance --
  connectLoop bumps the counter on every attempt in both the
  async-initial-connect and mid-stream reconnect phases, so a frozen
  counter means the loop exited silently. A plain attempts >= 1 check
  would not discriminate: pre-budget attempts would satisfy it.
  Applied to testAsyncNoServerRetriesForeverNoTerminal,
  testConnectionLostRetriesForeverNoTerminal, and
  testReconnectNeverGivesUpInvariantB.

- RECOVERY: in testReconnectNeverGivesUpInvariantB, revive a server on
  the SAME port (the requestedPort constructor exists for down-then-up
  outage realism) and require a replayed frame to arrive -- end-to-end
  proof that "retry forever" converts into delivery once the server
  returns. The revival handler is receipt-only: faking ACKs would
  couple the test to the exact replay FSN sequence and a wrong-seq ACK
  risks the invalid-ACK terminal.

Also fix the test comment claiming flush publishes to on-disk SF: no
sf_dir is configured there, so rows buffer in the in-RAM cursor ring.

Author: bluestreak01

core/src/test/java/io/questdb/client/test/cutlass/qwp/client/InitialConnectAsyncTest.java

@@ -139,6 +139,14 @@ public void testAsyncNoServerRetriesForeverNoTerminal() throws Exception {
             Assert.assertFalse(
                     "wasEverConnected() stays false while no server is reachable",
                     ((QwpWebSocketSender) sender).wasEverConnected());
+            // LIVENESS: no-terminal-in-window alone cannot distinguish
+            // "retries forever" from "gave up silently" -- an I/O thread
+            // that exits without latching a terminal also delivers nothing,
+            // and the flush above lands in SF regardless of wire liveness.
+            // The retry loop bumps getTotalReconnectAttempts() on every
+            // iteration, so the counter advancing NOW -- long past the
+            // (ignored) 200ms budget -- proves the loop is still running.
+            awaitReconnectAttemptsAdvance((QwpWebSocketSender) sender);
         } finally {
             sender.close();
         }
@@ -269,6 +277,11 @@ public void testConnectionLostRetriesForeverNoTerminal() throws Exception {
                 Assert.assertTrue(
                         "wasEverConnected() must remain true after the outage",
                         ((QwpWebSocketSender) sender).wasEverConnected());
+                // LIVENESS: same discriminator as the async-init test above.
+                // The mid-stream reconnect loop must still be making connect
+                // attempts long past the (ignored) 200ms budget -- not merely
+                // failing to report that it stopped.
+                awaitReconnectAttemptsAdvance((QwpWebSocketSender) sender);
             } finally {
                 // closeQuietly (not a bare close()) so a close-path exception
                 // cannot replace a pending AssertionError from the contract
@@ -321,6 +334,31 @@ private static void awaitAtLeastOneConnectAttempt(QwpWebSocketSender wss) {
         }
     }
 
+    /**
+     * Proves the I/O thread's retry loop is still ALIVE rather than merely
+     * quiet: snapshots {@link QwpWebSocketSender#getTotalReconnectAttempts()}
+     * and spins until it advances past the snapshot. Callers invoke this
+     * after the (ignored) reconnect budget has already elapsed, so a stuck
+     * counter means the loop gave up silently -- exactly the Invariant B
+     * regression a no-terminal-in-window assertion cannot see. A plain
+     * {@code attempts >= 1} check would NOT discriminate: attempts made
+     * before the budget expired would satisfy it even if the loop then
+     * exited.
+     */
+    private static void awaitReconnectAttemptsAdvance(QwpWebSocketSender wss) {
+        long snapshot = wss.getTotalReconnectAttempts();
+        long deadlineNanos = System.nanoTime() + TimeUnit.SECONDS.toNanos(5);
+        while (wss.getTotalReconnectAttempts() <= snapshot) {
+            if (System.nanoTime() > deadlineNanos) {
+                throw new AssertionError(
+                        "Invariant B violation: reconnect attempts stuck at "
+                                + snapshot + " for 5s past the budget -- the I/O "
+                                + "thread stopped retrying without surfacing a terminal");
+            }
+            io.questdb.client.std.Compat.onSpinWait();
+        }
+    }
+
     /**
      * Closes the sender, tolerating close-path teardown noise only. Used
      * instead of a broad {@code catch (Exception ignored)} around a whole

core/src/test/java/io/questdb/client/test/cutlass/qwp/client/ReconnectTest.java

@@ -104,10 +104,16 @@ public void testReconnectAfterServerInducedDisconnect() throws Exception {
     public void testReconnectNeverGivesUpInvariantB() throws Exception {
         // INVARIANT B: server is up at first (initial connect + ACK), then torn
         // down. The I/O loop enters reconnect and must retry FOREVER -- flush()
-        // must keep succeeding (publishing to on-disk SF), never surface a
-        // give-up / budget terminal. The rows are safe in SF and the server may
-        // return, so reconnect_max_duration_millis is ignored as a give-up
-        // deadline.
+        // must keep succeeding (publishing to the in-RAM cursor ring; no sf_dir
+        // is configured here), never surface a give-up / budget terminal. The
+        // rows are buffered and the server may return, so
+        // reconnect_max_duration_millis is ignored as a give-up deadline.
+        // Because flush() succeeds on the user thread regardless of I/O-thread
+        // liveness, the no-throw loop below is backed by two discriminators:
+        // the reconnect-attempt counter must keep advancing past the budget
+        // (loop is alive, not silently exited), and a server revived on the
+        // SAME port must receive the replayed frames (retrying converts into
+        // delivery).
         try (TestWebSocketServer server = new TestWebSocketServer(new AckHandler())) {
             int port = server.getPort();
             server.start();
@@ -144,6 +150,42 @@ public void testReconnectNeverGivesUpInvariantB() throws Exception {
                     }
                     Thread.sleep(50);
                 }
+
+                if (observed == null) {
+                    // LIVENESS: snapshot the attempt counter well past the
+                    // (ignored) 300ms budget and require it to advance. A
+                    // silently-exited I/O thread leaves it frozen -- the one
+                    // regression the flush loop above cannot see.
+                    io.questdb.client.cutlass.qwp.client.QwpWebSocketSender wss =
+                            (io.questdb.client.cutlass.qwp.client.QwpWebSocketSender) sender;
+                    long snapshot = wss.getTotalReconnectAttempts();
+                    long spinDeadline = System.currentTimeMillis() + 5_000;
+                    while (wss.getTotalReconnectAttempts() <= snapshot
+                            && System.currentTimeMillis() < spinDeadline) {
+                        Thread.sleep(20);
+                    }
+                    Assert.assertTrue(
+                            "Invariant B violation: reconnect attempts stuck at "
+                                    + snapshot + " for 5s past the (ignored) 300ms "
+                                    + "budget -- the I/O loop stopped retrying "
+                                    + "without surfacing a terminal",
+                            wss.getTotalReconnectAttempts() > snapshot);
+
+                    // RECOVERY: bring a server back on the SAME port (the
+                    // requestedPort constructor exists for down-then-up
+                    // outage realism). The still-retrying loop must reconnect
+                    // and replay the frames that accumulated unacked during
+                    // the outage -- a binary frame arriving on the revived
+                    // endpoint is end-to-end proof that "retry forever"
+                    // converts into delivery once the server returns.
+                    ReceiveOnlyHandler revived = new ReceiveOnlyHandler();
+                    try (TestWebSocketServer server2 =
+                                 new TestWebSocketServer(revived, false, null, port)) {
+                        server2.start();
+                        Assert.assertTrue(server2.awaitStart(5, TimeUnit.SECONDS));
+                        waitFor(() -> revived.totalBinaryReceived.get() >= 1, 5_000);
+                    }
+                }
             } finally {
                 try {
                     sender.close();
@@ -529,6 +571,22 @@ public void close() {
         }
     }
 
+    /**
+     * Counts binary frames without ACKing. Receipt-only revival probe for
+     * {@link #testReconnectNeverGivesUpInvariantB}: observing a replayed
+     * frame proves reconnect + replay without coupling the test to the
+     * exact FSN sequence a faked ACK would have to carry (a wrong-seq ACK
+     * risks tripping the invalid-ACK terminal and polluting the result).
+     */
+    private static class ReceiveOnlyHandler implements TestWebSocketServer.WebSocketServerHandler {
+        final AtomicLong totalBinaryReceived = new AtomicLong();
+
+        @Override
+        public void onBinaryMessage(TestWebSocketServer.ClientHandler client, byte[] data) {
+            totalBinaryReceived.incrementAndGet();
+        }
+    }
+
     /** Acks every binary frame so the sender doesn't hang. */
     private static class AckHandler implements TestWebSocketServer.WebSocketServerHandler {
         private final AtomicLong nextSeq = new AtomicLong(0);