Give the catch-up cap gap a settle budget

Two hardening fixes to the split/catch-up paths for heterogeneous /
rolling-cap clusters.

Catch-up cap-gap settle budget (M1): when a symbol-dict catch-up entry
is too large for the fresh server's advertised batch cap, sendDictCatchUp
latched a terminal on first sight. A homogeneous cluster never trips this
(an entry that fit its data frame under a cap always fits its bare
catch-up frame under that same cap), but a failover to a smaller-cap node
could kill the sender for an entry an earlier node accepted. It now
retries across MAX_CATCHUP_CAP_GAP_ATTEMPTS consecutive reconnects
(throwing a retriable CatchUpSendException, no recordFatal), riding out
the transient window until a larger-cap node returns; only a persistent
gap latches the terminal. A successful catch-up resets the budget, and a
transient reconnect never reaches the catch-up so it neither increments
nor burns it. Matches the orphan drainer's durable-ack settle budget.

Split-flush cap snapshot (M2): flushPendingRows now snapshots the
volatile serverMaxBatchSize once and threads it into flushPendingRowsSplit
for both the pre-flight sizing and the publish-loop assert. A mid-flush
failover (the I/O thread lowers the cap) could previously make the two
passes size against different caps, breaking the all-or-nothing guarantee
and firing the assert on a legitimate race. The next flush picks up the
new cap.

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

Author: glasstiger

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

@@ -3512,11 +3512,19 @@ private void flushPendingRows(boolean deferCommit) {
         int messageSize = encoder.finishMessage();
         QwpBufferWriter buffer = encoder.getBuffer();
 
-        if (serverMaxBatchSize > 0 && messageSize > serverMaxBatchSize) {
+        // Snapshot the volatile cap ONCE for this whole flush. The I/O thread lowers
+        // serverMaxBatchSize on a mid-stream failover to a smaller-cap node
+        // (applyServerBatchSizeLimit); if the split pre-flight and the publish loop
+        // re-read the field independently, a failover between them would size frames
+        // against two different caps -- breaking the all-or-nothing guarantee and
+        // firing the publish-loop assert on a legitimate race. Both use this snapshot;
+        // the next flush picks up the new cap.
+        int cap = serverMaxBatchSize;
+        if (cap > 0 && messageSize > cap) {
             // The combined frame's delta-entry bytes are byte-identical to the first
             // split frame's (same baseline + batch max), so capture the length now
             // for the arithmetic frame-sizing in flushPendingRowsSplit.
-            flushPendingRowsSplit(keys, deferCommit, encoder.getDeltaEntriesLen());
+            flushPendingRowsSplit(keys, deferCommit, encoder.getDeltaEntriesLen(), cap);
             return;
         }
 
@@ -3568,9 +3576,9 @@ private void flushPendingRows(boolean deferCommit) {
      *                    carry FLAG_DEFER_COMMIT. When false, only the
      *                    last message omits the flag.
      */
-    private void flushPendingRowsSplit(ObjList<CharSequence> keys, boolean deferCommit, int combinedDeltaEntriesLen) {
+    private void flushPendingRowsSplit(ObjList<CharSequence> keys, boolean deferCommit, int combinedDeltaEntriesLen, int cap) {
         if (LOG.isDebugEnabled()) {
-            LOG.debug("Splitting flush across multiple messages [serverMaxBatchSize={}, defer={}]", serverMaxBatchSize, deferCommit);
+            LOG.debug("Splitting flush across multiple messages [serverMaxBatchSize={}, defer={}]", cap, deferCommit);
         }
 
         // Collect non-empty table indices so we know which is last, AND pre-flight
@@ -3613,12 +3621,12 @@ private void flushPendingRowsSplit(ObjList<CharSequence> keys, boolean deferComm
                     + (deltaCount > 0 ? combinedDeltaEntriesLen : 0)
                     + splitFrameBodyBytes.getQuick(bodyIdx);
             bodyIdx++;
-            if (messageSize > serverMaxBatchSize) {
+            if (messageSize > cap) {
                 resetTableBuffersAfterFlush(keys);
                 throw new LineSenderException("single table batch too large for server batch cap")
                         .put(" [table=").put(tableName)
                         .put(", messageSize=").put(messageSize)
-                        .put(", serverMaxBatchSize=").put(serverMaxBatchSize).put(']');
+                        .put(", serverMaxBatchSize=").put(cap).put(']');
             }
             // Mirror advanceSentMaxSymbolId: once the first frame ships the batch's
             // new ids, the remaining frames carry an empty delta above the baseline.
@@ -3655,12 +3663,14 @@ private void flushPendingRowsSplit(ObjList<CharSequence> keys, boolean deferComm
             // The pre-flight pass above already verified every split frame fits the
             // cap, so none can be found oversized here -- which is what keeps this
             // loop from publishing (and stranding) a deferred prefix before an
-            // oversized table. The assert guards a future divergence between the two
-            // passes; it deliberately does NOT reset+throw here, because by this
-            // point a prefix may already be on the ring.
-            assert messageSize <= serverMaxBatchSize
+            // oversized table. Both passes size against the SAME snapshot cap, so a
+            // mid-flush failover cannot make them disagree; the assert therefore only
+            // catches a genuine divergence between the pre-flight arithmetic and the
+            // real encode (a future bug), not a cap race. It deliberately does NOT
+            // reset+throw here, because by this point a prefix may already be on the ring.
+            assert messageSize <= cap
                     : "split frame exceeded serverMaxBatchSize after pre-flight [table=" + tableName
-                    + ", messageSize=" + messageSize + ", serverMaxBatchSize=" + serverMaxBatchSize + ']';
+                    + ", messageSize=" + messageSize + ", serverMaxBatchSize=" + cap + ']';
 
             // Write-ahead persist before publish (see flushPendingRows). The
             // first split frame carries the batch's new symbols; the rest are

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

@@ -133,6 +133,18 @@ public final class CursorWebSocketSendLoop implements QuietCloseable {
      */
     public static final long DEFAULT_POISON_MIN_ESCALATION_WINDOW_MILLIS = 5_000L;
     private static final Logger LOG = LoggerFactory.getLogger(CursorWebSocketSendLoop.class);
+    // Settle budget for the symbol-dict catch-up cap gap: how many CONSECUTIVE
+    // reconnect attempts may find a single dictionary entry too large for the fresh
+    // server's advertised batch cap before the sender latches a terminal. A
+    // homogeneous cluster never trips it -- an entry that fit its data frame under a
+    // cap always fits its bare catch-up frame under that same cap -- so this only
+    // affects a heterogeneous / rolling-cap cluster, where a failover to a
+    // smaller-cap node can hit it for an entry an earlier node accepted. Retrying
+    // rides out the transient window until a larger-cap node returns; only a
+    // persistent gap (every reachable node too small for this many attempts) latches
+    // terminal, matching the orphan drainer's DEFAULT_MAX_DURABLE_ACK_MISMATCH_ATTEMPTS.
+    // A successful catch-up resets the counter (see sendDictCatchUp).
+    private static final int MAX_CATCHUP_CAP_GAP_ATTEMPTS = 16;
     // Hard ceiling for the lifetime-monotonic sent-dictionary mirror. The mirror
     // fields are int, so it cannot exceed Integer.MAX_VALUE bytes; reaching even
     // this needs ~200M+ distinct symbols on a single connection, far past any real
@@ -206,6 +218,12 @@ public final class CursorWebSocketSendLoop implements QuietCloseable {
     // for the connection's lifetime (a reconnect may need the whole dictionary at
     // any moment), so it cannot be dropped; it is an intentional cost of the feature.
     private final boolean deltaDictEnabled;
+    // Consecutive reconnect attempts whose symbol-dict catch-up found an entry too
+    // large for the fresh server's batch cap (see MAX_CATCHUP_CAP_GAP_ATTEMPTS). A
+    // successful catch-up resets it; it is 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;
     // 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 wire sequences (nextWireSeq), so nextWireSeq > 0 no longer implies
@@ -2243,24 +2261,36 @@ private int sendDictCatchUp() {
                     + NativeBufferWriter.varintSize(1)
                     + entryBytes;
             if (soloFrameLen > frameLimit) {
-                // Non-retriable: the entry will not shrink and the same cluster
-                // re-advertises the same cap, so reconnecting would livelock.
-                // Latch a terminal (the data must be resent after the cap is
-                // raised) rather than calling fail() -- which, from inside the
-                // catch-up, would re-enter connectLoop (see CatchUpSendException).
+                // Cap gap: this entry cannot be re-registered under the fresh
+                // server's advertised cap. A HOMOGENEOUS cluster never reaches here
+                // (an entry that fit its data frame under a cap always fits its bare
+                // catch-up frame under that same cap), so the only way in is a
+                // heterogeneous / rolling-cap failover to a smaller-cap node.
                 //
-                // Tradeoff (heterogeneous / rolling-cap clusters): a symbol
-                // accepted under a larger/absent cap can hit this on failover to a
-                // smaller-cap node, and the hard terminal does NOT self-recover if
-                // a later node advertises a larger cap -- the producer must be
-                // resumed after the data is resent (or the cap raised). Bounding
-                // symbol size at ingest, or a settle budget across reconnects
-                // before latching, would relax this, but both are larger changes;
-                // the terminal keeps the homogeneous common case livelock-free.
+                // Give the cluster a settle budget instead of latching on first
+                // sight: a larger-cap node may return, so retry across reconnects
+                // and only latch a terminal after MAX_CATCHUP_CAP_GAP_ATTEMPTS
+                // consecutive attempts still find it too large. Under budget the
+                // throw is RETRIABLE (no recordFatal) -- connectLoop reconnects with
+                // backoff and re-runs the catch-up, which resets the counter on a
+                // node that accepts it. A transient reconnect (connect/upgrade
+                // failure, role reject) never reaches the catch-up, so it neither
+                // increments nor burns this budget. On exhaustion latch via
+                // recordFatal, NOT fail() -- failing from inside the catch-up would
+                // re-enter connectLoop (see CatchUpSendException); the data must be
+                // resent after the cap is raised.
+                catchUpCapGapAttempts++;
+                boolean exhausted = catchUpCapGapAttempts >= MAX_CATCHUP_CAP_GAP_ATTEMPTS;
                 LineSenderException err = new LineSenderException(
                         "symbol dictionary entry too large for the server batch cap during catch-up ["
-                                + "frameLen=" + soloFrameLen + ", cap=" + cap + ']');
-                recordFatal(err);
+                                + "frameLen=" + soloFrameLen + ", cap=" + cap + ", attempt="
+                                + catchUpCapGapAttempts + '/' + MAX_CATCHUP_CAP_GAP_ATTEMPTS + ']'
+                                + (exhausted
+                                ? "; the data must be resent after the cap is raised"
+                                : "; retrying -- a larger-cap node may return"));
+                if (exhausted) {
+                    recordFatal(err);
+                }
                 throw new CatchUpSendException(err);
             }
             if (chunkSymbols > 0 && chunkBytes + entryBytes > budget) {
@@ -2279,6 +2309,9 @@ private int sendDictCatchUp() {
             sendCatchUpChunk(chunkStartId, chunkSymbols, chunkStartAddr, (int) chunkBytes);
             framesSent++;
         }
+        // The whole dictionary re-registered without a cap gap: this node accepts
+        // every entry, so the cap-gap episode (if any) is over -- reset the budget.
+        catchUpCapGapAttempts = 0;
         return framesSent;
     }
 

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

@@ -212,11 +212,15 @@ public void testFixedCapNearBoundarySymbolCatchesUpWithoutTerminal() throws Exce
     @Test
     public void testCatchUpEntryTooLargeForCapFailsTerminally() throws Exception {
         // A dictionary entry that exceeds the reconnect server's per-chunk budget
-        // (cap - HEADER_SIZE - 16) cannot be shipped as a catch-up chunk.
-        // sendDictCatchUp must latch a clean terminal ("... during catch-up")
-        // rather than call fail(): pre-fix the oversized entry drove an endless
-        // reconnect loop (the entry never shrinks and the same cluster
-        // re-advertises the same cap) and re-entered connectLoop from the catch-up.
+        // (cap - HEADER_SIZE - 16) cannot be shipped as a catch-up chunk. Every
+        // reachable node re-advertises the same small cap here, so the gap never
+        // resolves: sendDictCatchUp must retry across the settle budget and then
+        // latch a clean terminal ("... during catch-up") rather than call fail()
+        // (which from inside the catch-up re-enters connectLoop). Pre-fix it latched
+        // on the FIRST cap gap; the settle budget (MAX_CATCHUP_CAP_GAP_ATTEMPTS)
+        // rides out a transient smaller-cap window first (see the retry sibling in
+        // CursorWebSocketSendLoopCatchUpAlignmentTest), and only a persistent gap
+        // exhausts it. Small reconnect backoffs keep the budgeted attempts fast.
         //
         // Connection 1 advertises no cap, so the ~202-byte symbol registers and
         // enters the sent-dictionary mirror. The handler then shrinks the
@@ -234,7 +238,8 @@ public void testCatchUpEntryTooLargeForCapFailsTerminally() throws Exception {
 
                 String bigSymbol = TestUtils.repeat("x", 200); // ~202-byte dict entry
                 LineSenderException terminal = null;
-                Sender sender = Sender.fromConfig("ws::addr=localhost:" + port + ";");
+                Sender sender = Sender.fromConfig("ws::addr=localhost:" + port
+                        + ";reconnect_initial_backoff_millis=10;reconnect_max_backoff_millis=50;");
                 try {
                     sender.table("t").symbol("s", bigSymbol).longColumn("v", 1L).atNow();
                     sender.flush();

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

@@ -26,6 +26,7 @@
 
 import io.questdb.client.DefaultHttpClientConfiguration;
 import io.questdb.client.cutlass.http.client.WebSocketClient;
+import io.questdb.client.cutlass.line.LineSenderException;
 import io.questdb.client.cutlass.qwp.client.WebSocketResponse;
 import io.questdb.client.cutlass.qwp.client.sf.cursor.CursorSendEngine;
 import io.questdb.client.cutlass.qwp.client.sf.cursor.CursorWebSocketSendLoop;
@@ -265,6 +266,61 @@ public void testCatchUpChunkFrameSizeOverflowFailsLoud() throws Exception {
         });
     }
 
+    @Test
+    public void testCatchUpCapGapRetriesUntilBudgetThenLatches() throws Exception {
+        // M1: an entry too large for the fresh server's cap during catch-up (a
+        // heterogeneous / rolling-cap failover to a smaller-cap node) must NOT latch
+        // on first sight. sendDictCatchUp throws a RETRIABLE CatchUpSendException so
+        // the reconnect loop rides it out -- a larger-cap node may return -- and only
+        // after MAX_CATCHUP_CAP_GAP_ATTEMPTS consecutive cap gaps does it recordFatal.
+        // Pre-fix the first cap gap latched a terminal, so one transient failover to a
+        // smaller-cap node killed the sender. (A successful catch-up resets the budget;
+        // the other catch-up tests, which use a fitting cap, never trip it.)
+        TestUtils.assertMemoryLeak(() -> {
+            Field maxField = CursorWebSocketSendLoop.class.getDeclaredField("MAX_CATCHUP_CAP_GAP_ATTEMPTS");
+            maxField.setAccessible(true);
+            int maxAttempts = maxField.getInt(null);
+            // cap 160 => catch-up budget is below a ~216-byte solo frame for a 200-char symbol.
+            CatchUpCapturingClient client = new CatchUpCapturingClient(160);
+            try (CursorSendEngine engine = newEngine()) {
+                CursorWebSocketSendLoop loop = newLoop(engine, client);
+                try {
+                    seedMirror(loop, TestUtils.repeat("x", 200));
+                    // Attempts 1 .. max-1 are retriable: no terminal is latched.
+                    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());
+                            assertTrue("attempt " + i + " must name the catch-up cap gap: "
+                                            + e.getCause().getMessage(),
+                                    e.getCause().getMessage().contains("during catch-up"));
+                        }
+                        loop.checkError(); // under budget => retriable => no terminal
+                    }
+                    // The exhausting attempt still throws, and now latches the terminal.
+                    try {
+                        invokeSetWireBaselineWithCatchUp(loop, engine.ackedFsn() + 1L);
+                        fail("the exhausting cap gap must still raise CatchUpSendException");
+                    } catch (InvocationTargetException e) {
+                        assertEquals("CatchUpSendException", e.getCause().getClass().getSimpleName());
+                    }
+                    try {
+                        loop.checkError();
+                        fail("exhausting the cap-gap settle budget must latch a terminal");
+                    } 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();
+                }
+            }
+        });
+    }
+
     private static void appendFrames(CursorSendEngine engine, int count) {
         long buf = Unsafe.malloc(16, MemoryTag.NATIVE_DEFAULT);
         try {