fix(qwp): split role-reject observability out of onDurableAckUnavailable (M10)

Unreleased commits d0a794a/4200697 made onDurableAckUnavailable fire for
BOTH the durable-ack capability gap and the transient all-replica window,
passing whichever mode-local counter was current with no discriminator. A
listener written against the released 1.3.4 contract (capability-gap-only,
counter marching toward the 16-attempt escalation) saw a non-monotonic,
ambiguous stream: alerting keyed on "attemptNumber approaching 16"
false-positived on role-reject windows (which never escalate, Invariant B)
and could miss real budget pressure.

Split the streams before 1.3.5 ships the break:
- new default method onPrimaryUnavailable(slotPath, attemptNumber) carries
  the role-reject count (retried indefinitely, never escalates); the no-op
  default keeps 1.3.4 implementors source/binary compatible
- onDurableAckUnavailable reverts to the released capability-gap-only
  semantics; its episode counter still restarts on an intervening role
  reject (topology churn grants a fresh settle budget), but the reset is
  now attributable to the onPrimaryUnavailable delivery between episodes
- budget/episode/escalation logic untouched (Invariant B tests unmodified
  except for the per-stream assertion split)

Author: bluestreak01

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

@@ -274,9 +274,9 @@ public WebSocketClient connectWithDurableAckRetry() {
                 BackgroundDrainerListener l = listener;
                 if (l != null) {
                     try {
-                        l.onDurableAckUnavailable(slotPath, roleRejectAttempts);
+                        l.onPrimaryUnavailable(slotPath, roleRejectAttempts);
                     } catch (Throwable cb) {
-                        LOG.warn("drainer listener onDurableAckUnavailable threw: {}",
+                        LOG.warn("drainer listener onPrimaryUnavailable threw: {}",
                                 cb.getMessage());
                     }
                 }

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

@@ -57,20 +57,42 @@ public interface BackgroundDrainerListener {
     void onDurableAckPersistentFailure(String slotPath, int totalAttempts, long elapsedMillis);
 
     /**
-     * Fired when a connect sweep found durable ack unavailable — either a
-     * genuine capability gap ({@code QwpDurableAckMismatchException}: an
-     * endpoint upgrades but does not advertise durable ack) or a transient
-     * all-replica failover window (role reject). The drainer will back off
-     * and retry; this callback is purely observability. Source data stays
-     * pinned regardless because the loop runs in {@code durableAckMode=true}
-     * and only trims on STATUS_DURABLE_ACK.
+     * Fired when a connect sweep hit a genuine durable-ack capability gap
+     * ({@code QwpDurableAckMismatchException}: an endpoint upgrades but does
+     * not advertise durable ack). The drainer will back off and retry within
+     * its settle budget; this callback is purely observability. Source data
+     * stays pinned regardless because the loop runs in
+     * {@code durableAckMode=true} and only trims on STATUS_DURABLE_ACK.
+     * A transient all-replica failover window (role reject) never fires this
+     * callback — it is surfaced through {@link #onPrimaryUnavailable}.
      *
      * @param slotPath      slot the drainer is processing
-     * @param attemptNumber 1-based attempt count within the current mode:
-     *                      the running role-reject count for a transient
-     *                      window, or the attempt number within the current
-     *                      capability-gap episode (restarts when a role
-     *                      reject resets the episode)
+     * @param attemptNumber 1-based attempt number within the current
+     *                      capability-gap EPISODE. The counter restarts when
+     *                      an intervening role reject resets the episode —
+     *                      topology churn grants the next gap a fresh settle
+     *                      budget, which is correct behavior — and with the
+     *                      streams separated the reset's cause is visible as
+     *                      an {@link #onPrimaryUnavailable} delivery between
+     *                      the two episodes
      */
     void onDurableAckUnavailable(String slotPath, int attemptNumber);
+
+    /**
+     * Fired when a connect sweep found every reachable endpoint to be a
+     * REPLICA — a transient all-replica failover window (role reject). A
+     * replica is promotable and a primary will reappear, so the drainer
+     * retries indefinitely under Invariant B: this condition NEVER escalates
+     * and is never followed by {@link #onDurableAckPersistentFailure}. Runs
+     * on the drainer thread; implementations must not block. The no-op
+     * default keeps every implementor of the released 1.3.4 contract source-
+     * and binary-compatible.
+     *
+     * @param slotPath      slot the drainer is processing
+     * @param attemptNumber 1-based running role-reject count within the
+     *                      current connect loop (resets across connect
+     *                      re-entries)
+     */
+    default void onPrimaryUnavailable(String slotPath, int attemptNumber) {
+    }
 }

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

@@ -44,6 +44,7 @@
 
 import java.nio.file.Paths;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
 import java.util.concurrent.CountDownLatch;
@@ -562,14 +563,19 @@ public void testRoleRejectChurnDoesNotConsumeCapabilityGapBudgetInvariantB() thr
                     cap, listener.lastPersistentTotalAttempts.get());
             assertEquals("full settle budget must be granted after the transient window",
                     roleRejects + cap, factory.attempts());
-            // Per-mode attempt numbering: 1..20 for the transient window, then a
-            // fresh 1..15 for the gap episode (the 16th fires persistent-failure).
-            assertEquals(roleRejects + cap - 1, listener.unavailableAttempts.size());
+            // M10 split: the transient all-replica window lands on the
+            // onPrimaryUnavailable stream (1..20), the capability-gap episode
+            // on onDurableAckUnavailable (1..15; the 16th fires
+            // persistent-failure instead). Neither stream sees the other's
+            // counter, so a listener alerting on "attemptNumber approaching
+            // the cap" no longer false-positives on role-reject churn.
+            assertEquals(roleRejects, listener.primaryUnavailableAttempts.size());
             for (int i = 0; i < roleRejects; i++) {
-                assertEquals(Integer.valueOf(i + 1), listener.unavailableAttempts.get(i));
+                assertEquals(Integer.valueOf(i + 1), listener.primaryUnavailableAttempts.get(i));
             }
+            assertEquals(cap - 1, listener.unavailableAttempts.size());
             for (int i = 0; i < cap - 1; i++) {
-                assertEquals(Integer.valueOf(i + 1), listener.unavailableAttempts.get(roleRejects + i));
+                assertEquals(Integer.valueOf(i + 1), listener.unavailableAttempts.get(i));
             }
             assertTrue(Files.exists(slotPath + "/" + OrphanScanner.FAILED_SENTINEL_NAME));
         });
@@ -646,10 +652,62 @@ public void testRoleRejectResetsCapabilityGapEpisode() throws Exception {
                     cap, listener.lastPersistentTotalAttempts.get());
             // 15 gap + 1 role reject + 16 gap = 32 sweeps total.
             assertEquals(2 * cap, factory.attempts());
+            // M10 split, per-stream: the DA stream carries both episodes'
+            // per-episode numbering (1..15, then 1..15 again -- the second
+            // episode's 16th attempt fires persistent-failure instead), and
+            // the reset between them is attributable: exactly one role reject
+            // on the primary stream. Before the split the reset was an
+            // ambiguous non-monotonic drop in a single stream.
+            List<Integer> expectedDaStream = new ArrayList<>();
+            for (int episode = 0; episode < 2; episode++) {
+                for (int i = 1; i <= cap - 1; i++) {
+                    expectedDaStream.add(i);
+                }
+            }
+            assertEquals(expectedDaStream, listener.unavailableAttempts);
+            assertEquals(Collections.singletonList(1), listener.primaryUnavailableAttempts);
             assertTrue(Files.exists(slotPath + "/" + OrphanScanner.FAILED_SENTINEL_NAME));
         });
     }
 
+    @Test
+    public void testRoleRejectAndCapabilityGapLandOnSeparateStreams() throws Exception {
+        assertMemoryLeak(() -> {
+            // M10 discriminator: gap -> role reject -> gap -> success. The
+            // released 1.3.4 contract fed BOTH conditions to
+            // onDurableAckUnavailable, so this script produced the ambiguous
+            // stream [1, 1, 1] -- a listener could not tell a budget-bound
+            // capability-gap episode from a never-escalating role-reject
+            // window, and could not see WHY the episode counter reset. With
+            // the split, the DA stream carries only the two one-attempt gap
+            // episodes ([1, 1] -- the reset stays visible) and the role
+            // reject that caused the reset lands on the primary stream ([1]).
+            CountingListener listener = new CountingListener();
+            AtomicInteger sweeps = new AtomicInteger();
+            ScriptedFactory factory = ScriptedFactory.failingTimes(3, () -> {
+                if (sweeps.incrementAndGet() == 2) {
+                    return new QwpIngressRoleRejectedException(
+                            QwpIngressRoleRejectedException.ROLE_REPLICA, "127.0.0.1", 9000);
+                }
+                return new QwpDurableAckMismatchException("h", 1234, "primary");
+            });
+            BackgroundDrainer drainer = newDrainer(factory);
+            drainer.setListener(listener);
+            WebSocketClient out = drainer.connectWithDurableAckRetry();
+            assertSame(factory.successSentinel(), out);
+            assertEquals(4, factory.attempts());
+            assertEquals("DA stream must carry only the gap episodes, each"
+                            + " restarting at 1 after the role-reject reset",
+                    Arrays.asList(1, 1), listener.unavailableAttempts);
+            assertEquals("role reject must land on the primary stream",
+                    Collections.singletonList(1), listener.primaryUnavailableAttempts);
+            assertEquals(Collections.singletonList(slotPath), listener.primaryUnavailableSlotPaths);
+            assertEquals(BackgroundDrainer.DrainOutcome.PENDING, drainer.outcome());
+            assertEquals(0, listener.persistentFailures.get());
+            assertFalse(Files.exists(slotPath + "/" + OrphanScanner.FAILED_SENTINEL_NAME));
+        });
+    }
+
     @Test
     public void testTransportErrorDoesNotResetCapabilityGapEpisode() throws Exception {
         assertMemoryLeak(() -> {
@@ -788,9 +846,11 @@ public void testRoleRejectGrantsFreshWallClockToNextGapEpisode() {
                 0, listener.persistentFailures.get());
         assertFalse("no .failed sentinel: both gap episodes stayed inside their budgets",
                 Files.exists(slotPath + "/" + OrphanScanner.FAILED_SENTINEL_NAME));
-        // Per-mode attempt numbering across the reset: gaps 1,2 -- role reject
-        // 1 -- fresh gap episode 1,2.
-        assertEquals(java.util.Arrays.asList(1, 2, 1, 1, 2), listener.unavailableAttempts);
+        // Per-stream attempt numbering across the reset (M10 split): the DA
+        // stream carries gaps 1,2 then the fresh episode's 1,2; the role
+        // reject that restarted the episode lands on the primary stream.
+        assertEquals(Arrays.asList(1, 2, 1, 2), listener.unavailableAttempts);
+        assertEquals(Collections.singletonList(1), listener.primaryUnavailableAttempts);
     }
 
     @Test
@@ -907,6 +967,8 @@ private static final class CountingListener implements BackgroundDrainerListener
         final AtomicInteger lastPersistentElapsedMs = new AtomicInteger(-1);
         final AtomicInteger lastPersistentTotalAttempts = new AtomicInteger(-1);
         final AtomicInteger persistentFailures = new AtomicInteger();
+        final List<Integer> primaryUnavailableAttempts = new ArrayList<>();
+        final List<String> primaryUnavailableSlotPaths = new ArrayList<>();
         final List<Integer> unavailableAttempts = new ArrayList<>();
         final List<String> unavailableSlotPaths = new ArrayList<>();
 
@@ -922,6 +984,12 @@ public synchronized void onDurableAckUnavailable(String slotPath, int attemptNum
             unavailableSlotPaths.add(slotPath);
             unavailableAttempts.add(attemptNumber);
         }
+
+        @Override
+        public synchronized void onPrimaryUnavailable(String slotPath, int attemptNumber) {
+            primaryUnavailableSlotPaths.add(slotPath);
+            primaryUnavailableAttempts.add(attemptNumber);
+        }
     }
 
     /**