Fix catch-up terminal on a homogeneous batch cap

The reconnect symbol-dictionary catch-up latched a terminal for any
entry exceeding the packing budget (cap - HEADER_SIZE - 16). That
reserve is larger than the minimal data-frame overhead, so a symbol the
producer had already shipped under a given cap could exceed the budget
and trip the terminal on reconnect -- permanently hard-failing a running
producer on a homogeneous single-cap cluster, the exact failure the
store-and-forward path exists to survive.

sendDictCatchUp now tests the actual solo catch-up frame (header + the
entry's delta varints + the entry bytes) against the cap. That frame is
always smaller than the data frame that already carried the entry, so an
accepted symbol always re-registers on the same cap; a genuinely
oversized entry on a shrunk (heterogeneous) cap still terminates. The
conservative packing budget stays, used only for multi-entry chunk
splitting.

Add a fixed-cap regression test: a 173-char symbol under a fixed cap of
200 is accepted, then re-registers gap-free across a reconnect. It fails
before this change, where the reverted condition terminates a frame that
fits the 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/sf/cursor/CursorWebSocketSendLoop.java

@@ -2171,12 +2171,21 @@ private long readVarintAt(long p, long limit) {
      */
     private int sendDictCatchUp() {
         int cap = client.getServerMaxBatchSize();
-        // Symbol-bytes budget per frame, leaving room for the 12-byte header and
-        // the two delta-section varints. cap <= 0 means the server advertised no
-        // limit -- send the whole dictionary in one frame, but still bound the
-        // budget by MAX_SENT_DICT_BYTES so sendCatchUpChunk's int frameLen
-        // (HEADER_SIZE + varints + symbolsLen) cannot overflow on a pathological
-        // multi-GB dictionary (unreachable at real cardinality; defensive).
+        // The frame ceiling a catch-up chunk must not exceed: the server's
+        // advertised cap, or -- when the server advertises none (cap <= 0) --
+        // MAX_SENT_DICT_BYTES so sendCatchUpChunk's int frameLen (HEADER_SIZE +
+        // varints + symbolsLen) cannot overflow on a pathological multi-GB
+        // dictionary (unreachable at real cardinality; defensive). Used by the
+        // single-entry terminal below, which measures the real solo frame.
+        int frameLimit = cap > 0 ? cap : MAX_SENT_DICT_BYTES;
+        // Symbol-bytes budget for PACKING several entries into one chunk, leaving
+        // room for the 12-byte header and the two delta-section varints. Kept
+        // deliberately conservative (reserving 16 for the varints): it only makes a
+        // multi-entry chunk split marginally earlier, never over the cap. It must
+        // NOT gate the single-entry terminal -- that reserve is larger than the
+        // minimal data-frame overhead, so an entry the producer already shipped
+        // under this cap could exceed the reserve yet still fit its own catch-up
+        // frame; the terminal tests the real solo frame against frameLimit instead.
         int budget = cap > 0
                 ? Math.max(1, cap - QwpConstants.HEADER_SIZE - 16)
                 : MAX_SENT_DICT_BYTES - QwpConstants.HEADER_SIZE - 16;
@@ -2192,7 +2201,20 @@ private int sendDictCatchUp() {
             long len = readVarintAt(p, limit); // entry length prefix; varintEnd -> just past prefix
             long entryEnd = varintEnd + len;   // just past [len varint][utf8 bytes]
             long entryBytes = entryEnd - entryStart;
-            if (entryBytes > budget) {
+            // The exact table-less frame sendCatchUpChunk would build for THIS entry
+            // alone: header + deltaStart varint (the entry's own global id) +
+            // deltaCount varint (1) + the entry bytes. Terminal only when even that
+            // solo frame exceeds the cap -- i.e. the entry genuinely cannot be
+            // re-registered. Testing the real solo frame (not the conservative
+            // packing budget above) is what keeps a HOMOGENEOUS cluster
+            // livelock-free: an entry the producer already shipped in a data frame
+            // under this cap (header + delta varints + entry + schema + >=1 row) is
+            // strictly larger than its bare catch-up frame, so it always fits here.
+            long soloFrameLen = QwpConstants.HEADER_SIZE
+                    + NativeBufferWriter.varintSize(chunkStartId + chunkSymbols)
+                    + 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
@@ -2209,7 +2231,7 @@ private int sendDictCatchUp() {
                 // the terminal keeps the homogeneous common case livelock-free.
                 LineSenderException err = new LineSenderException(
                         "symbol dictionary entry too large for the server batch cap during catch-up ["
-                                + "entryBytes=" + entryBytes + ", budget=" + budget + ']');
+                                + "frameLen=" + soloFrameLen + ", cap=" + cap + ']');
                 recordFatal(err);
                 throw new CatchUpSendException(err);
             }

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

@@ -99,6 +99,67 @@ public void testReconnectCatchUpRebuildsDictionary() throws Exception {
         });
     }
 
+    @Test
+    public void testFixedCapNearBoundarySymbolCatchesUpWithoutTerminal() throws Exception {
+        // Regression (homogeneous single cap): a symbol whose length sits just below
+        // the advertised cap is ACCEPTED into a data frame (messageSize <= cap) and
+        // enters the sent-dictionary mirror. On reconnect the catch-up must
+        // re-register it under the SAME cap -- the bare catch-up frame (header + two
+        // varints + the entry) is smaller than the data frame that already shipped
+        // it (which also carried the table schema + a row), so it fits.
+        //
+        // Pre-fix, the single-entry terminal used the conservative PACKING budget
+        // (cap - HEADER_SIZE - 16), which is stricter than the producer's publish
+        // gate (messageSize <= cap) by more than the minimal data-frame overhead. So
+        // a symbol accepted onto the wire under cap C could exceed that budget and
+        // trip a spurious "during catch-up" terminal, permanently hard-failing a
+        // running producer on its first transient reconnect. Concretely at cap=200:
+        // table("t").symbol("s", <173 chars>).atNow() encodes to 198 bytes (<=200,
+        // accepted), its dict entry is 2+173=175 bytes (> old budget 172 -> old
+        // terminal), while the real solo catch-up frame is 12+1+1+175=189 (<=200 ->
+        // fits). Unlike testCatchUpEntryTooLargeForCapFailsTerminally (a genuinely
+        // oversized entry on a shrunk cap, which MUST still terminate), this entry
+        // is legally shippable and must NOT terminate.
+        final int cap = 200;
+        final String nearCapSymbol = TestUtils.repeat("x", 173);
+        assertMemoryLeak(() -> {
+            CatchUpHandler handler = new CatchUpHandler();
+            try (TestWebSocketServer server = new TestWebSocketServer(handler)) {
+                server.setAdvertisedMaxBatchSize(cap); // same cap on every handshake (homogeneous)
+                int port = server.getPort();
+                server.start();
+                Assert.assertTrue(server.awaitStart(5, TimeUnit.SECONDS));
+
+                try (Sender sender = Sender.fromConfig("ws::addr=localhost:" + port + ";")) {
+                    // Symbol-only row so the near-cap symbol drives the frame size.
+                    sender.table("t").symbol("s", nearCapSymbol).atNow();
+                    sender.flush();
+                    waitFor(() -> handler.dictFor(1).size() >= 1, 5_000);
+                    waitFor(() -> handler.conn1Closed, 5_000);
+
+                    // The reconnect runs the catch-up under the SAME cap. Pre-fix
+                    // this latched a terminal (surfacing on this flush); post-fix the
+                    // catch-up ships the near-cap symbol and the flush goes through.
+                    sender.table("t").symbol("s", "beta").atNow();
+                    sender.flush();
+                    waitFor(() -> handler.connectionsAccepted.get() >= 2
+                            && handler.dictFor(2).size() >= 2, 5_000);
+                }
+
+                // Connection 2's dictionary, rebuilt purely from the frames it
+                // received, must hold the near-cap symbol (re-registered by the
+                // catch-up) and beta, gap-free -- proving the catch-up SHIPPED rather
+                // than terminated.
+                List<String> conn2 = handler.dictFor(2);
+                Assert.assertTrue("2nd connection saw a catch-up frame with 0 tables",
+                        handler.sawZeroTableFrameOnConn2);
+                Assert.assertEquals("2nd connection dictionary size", 2, conn2.size());
+                Assert.assertEquals(nearCapSymbol, conn2.get(0));
+                Assert.assertEquals("beta", conn2.get(1));
+            }
+        });
+    }
+
     @Test
     public void testCatchUpEntryTooLargeForCapFailsTerminally() throws Exception {
         // A dictionary entry that exceeds the reconnect server's per-chunk budget