Don't key the cap-gap episode off nanoTime's sign

sendDictCatchUp tested `catchUpCapGapFirstNanos < 0` to decide whether a
symbol-dictionary cap-gap episode was already open. A System.nanoTime()
instant is only meaningful as a difference: its origin is arbitrary and
the spec permits negative values, so its sign cannot encode state.

CursorWebSocketSendLoopCatchUpAlignmentTest backdates the episode anchor
by two hours to prove that a cap gap outliving the escalation window
latches a terminal. On Linux nanoTime() is nanos-since-boot, so on a
freshly booted CI agent it sits far below two hours of nanos and the
backdated anchor came out negative. The loop then read that real anchor
as "unset", re-anchored it to now on every strike and pinned
episodeNanos at ~0: the dwell was never satisfied, no terminal latched,
and the test failed. It fails on every machine with under two hours of
uptime and on no long-lived dev box, which is why only CI saw it.

Key the episode off the strike count instead, exactly as
recordHeadRejectionStrike keys its own off poisonStrikes. -1 survives as
a debug marker but no longer decides anything, so no state rides on a
timestamp's sign.

Add testCapGapEpisodeWithANegativeAnchorStillEscalates, which plants the
negative anchor directly and so pins the sentinel whatever the host's
uptime; the existing test only reddens where uptime happens to fall under
its two-hour backdate.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: glasstiger

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

@@ -301,15 +301,19 @@ public final class CursorWebSocketSendLoop implements QuietCloseable {
     // nor resets it. NOT reset per connection -- it measures the cap-gap episode
     // across reconnects so a persistent gap eventually latches. I/O-thread-only.
     private int catchUpCapGapAttempts;
-    // System.nanoTime() of the FIRST cap gap of the current episode, or -1 when no
-    // episode is open. Anchors the escalation dwell. Reset together with
-    // catchUpCapGapAttempts on a successful catch-up, so a node that accepts the
-    // dictionary ends the episode outright. Anchoring at the FIRST gap (not at loop
-    // entry, and not refreshed per attempt) is what keeps a TRANSIENT from burning the
-    // terminal budget: a transient never reaches the catch-up, so it neither increments
-    // the counter nor moves the anchor -- it can only lengthen the wall clock, which
-    // alone can never latch the terminal because the strike count still has to be
-    // satisfied. I/O-thread-only, like catchUpCapGapAttempts.
+    // System.nanoTime() of the FIRST cap gap of the current episode. Anchors the
+    // escalation dwell. Reset together with catchUpCapGapAttempts on a successful
+    // catch-up, so a node that accepts the dictionary ends the episode outright.
+    // Anchoring at the FIRST gap (not at loop entry, and not refreshed per attempt) is
+    // what keeps a TRANSIENT from burning the terminal budget: a transient never reaches
+    // the catch-up, so it neither increments the counter nor moves the anchor -- it can
+    // only lengthen the wall clock, which alone can never latch the terminal because the
+    // strike count still has to be satisfied. I/O-thread-only, like catchUpCapGapAttempts.
+    //
+    // -1 marks "no episode open" for a debugger or a heap dump, but it is NOT the test
+    // for one -- catchUpCapGapAttempts == 0 is (see sendDictCatchUp). A nanoTime instant
+    // is only meaningful as a difference: its origin is arbitrary and the spec permits
+    // negative values, so no state may ride on this field's sign.
     private long catchUpCapGapFirstNanos = -1L;
     // True once a real ring frame (data or commit) has been sent on the CURRENT
     // connection, as opposed to only the dictionary catch-up. The catch-up consumes
@@ -2442,8 +2446,18 @@ private int sendDictCatchUp() {
                 // an ordinary rolling restart of the larger-cap node. Escalating on the
                 // count alone would therefore hard-fail a live store-and-forward
                 // producer during a routine cluster operation.
+                //
+                // The STRIKE COUNT -- never the anchor's sign -- says whether an episode is
+                // already open. A System.nanoTime() instant is only meaningful as a
+                // difference: its origin is arbitrary and the spec permits negative values,
+                // so a sign test cannot tell an unset anchor from a real one. Wherever it
+                // misreads a real anchor as unset it re-anchors to now on EVERY strike,
+                // pinning episodeNanos at ~0, and the dwell can then never be satisfied --
+                // the terminal never latches, however long the gap truly persists.
+                // recordHeadRejectionStrike keys its episode off poisonStrikes for the same
+                // reason.
                 long nowNanos = System.nanoTime();
-                if (catchUpCapGapFirstNanos < 0) {
+                if (catchUpCapGapAttempts == 0) {
                     catchUpCapGapFirstNanos = nowNanos;
                 }
                 catchUpCapGapAttempts++;

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

@@ -330,6 +330,71 @@ public void testCatchUpCapGapStrikesAloneDoNotLatchWithinTheEscalationWindow() t
         });
     }
 
+    @Test
+    public void testCapGapEpisodeWithANegativeAnchorStillEscalates() throws Exception {
+        // A cap-gap episode anchored at a NEGATIVE nanoTime instant must escalate like any
+        // other. A System.nanoTime() value is only meaningful as a difference -- its origin
+        // is arbitrary and the spec permits negative values -- so no state may ride on the
+        // anchor's sign. sendDictCatchUp once tested catchUpCapGapFirstNanos < 0 to mean "no
+        // episode open": that read a negative anchor as unset, re-anchored it to now on every
+        // strike and pinned episodeNanos at ~0, so the dwell was never satisfied and the
+        // terminal could never latch, however long the cap gap truly persisted.
+        //
+        // That is what reddened CI. The sibling test above backdates the anchor two hours,
+        // and on Linux nanoTime() is nanos-since-boot: on a CI agent up ten minutes it is
+        // ~6e11, so "two hours ago" comes out ~ -6.6e12 -- negative. The defect therefore
+        // only surfaced where uptime is under that backdate: every fresh CI agent, and never
+        // a long-lived dev box (which is why it passed locally). Planting the negative anchor
+        // directly pins the sentinel on ANY machine, whatever its uptime.
+        TestUtils.assertMemoryLeak(() -> {
+            Field maxField = CursorWebSocketSendLoop.class.getDeclaredField("MAX_CATCHUP_CAP_GAP_ATTEMPTS");
+            maxField.setAccessible(true);
+            int maxAttempts = maxField.getInt(null);
+            CatchUpCapturingClient client = new CatchUpCapturingClient(160);
+            try (CursorSendEngine engine = newEngine()) {
+                // A one-hour dwell, against an anchor two hours back: satisfied on elapsed,
+                // but only if the negative anchor survives to the subtraction.
+                CursorWebSocketSendLoop loop = newLoop(engine, client, 3_600_000L);
+                try {
+                    seedMirror(loop, TestUtils.repeat("x", 200));
+                    // Satisfy the strike half of the AND, one short of the budget.
+                    for (int i = 1; i < maxAttempts; i++) {
+                        try {
+                            invokeSetWireBaselineWithCatchUp(loop, engine.ackedFsn() + 1L);
+                            fail("cap gap must raise a retriable CatchUpSendException (attempt " + i + ')');
+                        } catch (InvocationTargetException e) {
+                            assertEquals("CatchUpSendException", e.getCause().getClass().getSimpleName());
+                        }
+                        loop.checkError(); // dwell unmet => retriable, whatever the count
+                    }
+                    // The episode began two hours ago on a machine booted minutes ago.
+                    Field anchor = CursorWebSocketSendLoop.class.getDeclaredField("catchUpCapGapFirstNanos");
+                    anchor.setAccessible(true);
+                    anchor.setLong(loop, -TimeUnit.HOURS.toNanos(2));
+
+                    // Both halves now hold, so this strike must latch the terminal.
+                    try {
+                        invokeSetWireBaselineWithCatchUp(loop, engine.ackedFsn() + 1L);
+                        fail("the escalating cap gap must still raise CatchUpSendException");
+                    } catch (InvocationTargetException e) {
+                        assertEquals("CatchUpSendException", e.getCause().getClass().getSimpleName());
+                    }
+                    try {
+                        loop.checkError();
+                        fail("a cap-gap episode anchored at a negative nanoTime instant must still "
+                                + "escalate -- the anchor's sign carries no meaning");
+                    } catch (LineSenderException terminal) {
+                        assertTrue("terminal must name the exhausted catch-up cap gap: " + terminal.getMessage(),
+                                terminal.getMessage().contains("during catch-up")
+                                        && terminal.getMessage().contains("must be resent"));
+                    }
+                } finally {
+                    loop.close();
+                }
+            }
+        });
+    }
+
     @Test
     public void testTransientCatchUpFailureDoesNotBurnTheCapGapBudget() throws Exception {
         // A TRANSIENT catch-up failure (the wire drops mid-catch-up -- a flapping LB, a