Stop split-flush stranding a deferred prefix

glasstiger · claude · glasstiger · commit c065ad40c97d · 2026-07-10T23:46:36.000+01:00
flushPendingRowsSplit published each table's frame one at a time
(deferred, uncommitted) and only checked a frame against the server
batch cap after encoding it. An oversized non-first table therefore
threw after an earlier table's frame was already on the ring, where a
later commit delivered it as a partial batch -- while
resetTableBuffersAfterFlush discarded every source row. The caller saw
a failure for a batch that had partly committed.

Pre-flight every split frame's size before publishing any of them, so
the split is all-or-nothing: either every frame fits and all publish,
or none publish and it throws with nothing stranded. simBaseline
mirrors advanceSentMaxSymbolId, so a measured size equals the frame the
publish loop builds; the pass is read-only on the table buffers and
touches no delta/persist state. The cost is a second encode pass over
the split batch, already the exceptional large-batch path.

testOversizedTableSplitStrandsNothing reproduces the stranding -- it
fails on the old code with the earlier frame reaching the server -- and
passes with the pre-flight.

Also add testReconnectPreservesMonotonicDeltaBaseline: it pins that the
producer's sent-symbol watermark survives a reconnect, asserting the
first post-reconnect data frame carries a delta above the baseline
(deltaStart &gt;= 1) rather than re-shipping the whole dictionary from 0.
The existing catch-up tests assert only that the final dictionary is
complete, which a reset-then-redefine also satisfies.

Co-Authored-By: Claude Opus 4.8 (1M context) &lt;noreply@anthropic.com&gt;
diff --git a/core/src/main/java/io/questdb/client/cutlass/qwp/client/QwpWebSocketSender.java b/core/src/main/java/io/questdb/client/cutlass/qwp/client/QwpWebSocketSender.java
@@ -3552,16 +3552,48 @@ private void flushPendingRowsSplit(ObjList<CharSequence> keys, boolean deferComm
             LOG.debug("Splitting flush across multiple messages [serverMaxBatchSize={}, defer={}]", serverMaxBatchSize, deferCommit);
         }
 
-        // Collect non-empty table indices so we know which is last.
+        // Collect non-empty table indices so we know which is last, AND pre-flight
+        // every split frame's size BEFORE publishing any of them. The split hands
+        // frames to the ring one at a time (all but the last deferred -- appended but
+        // uncommitted); if a later table's frame were only found oversized
+        // mid-publish, the already-published prefix would strand on the ring, a
+        // subsequent commit would deliver it as a partial batch, and
+        // resetTableBuffersAfterFlush would discard every source row -- a partial
+        // commit the caller was told (by the throw) had failed. Checking all sizes up
+        // front makes the split all-or-nothing: either every frame fits and all
+        // publish, or none publish and we throw with nothing stranded. The cost is a
+        // second encode pass over the split batch, which is already the exceptional
+        // large-batch path. encode is read-only on the table buffer, and simBaseline
+        // mirrors the publish loop's baseline advance (advanceSentMaxSymbolId), so
+        // each measured size equals the frame the publish loop will build; this pass
+        // mutates no delta/persist state (the defer-commit flag is a header bit that
+        // does not change frame size).
         int nonEmptyCount = 0;
+        int simBaseline = symbolDeltaBaseline();
         for (int i = 0, n = keys.size(); i < n; i++) {
             CharSequence tableName = keys.getQuick(i);
             if (tableName == null) {
                 continue;
             }
             QwpTableBuffer tableBuffer = tableBuffers.get(tableName);
-            if (tableBuffer != null && tableBuffer.getRowCount() > 0) {
-                nonEmptyCount++;
+            if (tableBuffer == null || tableBuffer.getRowCount() == 0) {
+                continue;
+            }
+            nonEmptyCount++;
+            encoder.beginMessage(1, globalSymbolDictionary, simBaseline, currentBatchMaxSymbolId);
+            encoder.addTable(tableBuffer);
+            int messageSize = encoder.finishMessage();
+            if (messageSize > serverMaxBatchSize) {
+                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(']');
+            }
+            // Mirror advanceSentMaxSymbolId: once the first frame ships the batch's
+            // new ids, the remaining frames carry an empty delta above the baseline.
+            if (deltaDictEnabled && currentBatchMaxSymbolId > simBaseline) {
+                simBaseline = currentBatchMaxSymbolId;
             }
         }
 
@@ -3590,14 +3622,15 @@ private void flushPendingRowsSplit(ObjList<CharSequence> keys, boolean deferComm
             encoder.addTable(tableBuffer);
             int messageSize = encoder.finishMessage();
             QwpBufferWriter buffer = encoder.getBuffer();
-
-            if (messageSize > serverMaxBatchSize) {
-                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(']');
-            }
+            // 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
+                    : "split frame exceeded serverMaxBatchSize after pre-flight [table=" + tableName
+                    + ", messageSize=" + messageSize + ", serverMaxBatchSize=" + serverMaxBatchSize + ']';
 
             // Write-ahead persist before publish (see flushPendingRows). The
             // first split frame carries the batch's new symbols; the rest are
diff --git a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/DeltaDictCatchUpTest.java b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/DeltaDictCatchUpTest.java
@@ -102,6 +102,52 @@ public void testReconnectCatchUpRebuildsDictionary() throws Exception {
         });
     }
 
+    @Test
+    public void testReconnectPreservesMonotonicDeltaBaseline() throws Exception {
+        // Regression: the producer's sent-symbol watermark (sentMaxSymbolId) must
+        // SURVIVE a reconnect. resetSymbolDictStateForNewConnection deliberately
+        // leaves it untouched -- the I/O thread re-registers the whole dictionary via
+        // a catch-up frame before replay, so the producer keeps shipping deltas ABOVE
+        // the baseline across the wire boundary. If a regression reset it on
+        // reconnect, the first post-reconnect data frame would re-ship the whole
+        // dictionary inline (deltaStart=0), pure wasted bandwidth. The sibling
+        // testReconnectCatchUpRebuildsDictionary asserts only that the final
+        // dictionary is complete -- which a reset-then-redefine ALSO satisfies (the
+        // server tolerates the redefinition) -- so it does NOT catch that regression.
+        // This pins the baseline survival directly: connection 2's data frame must
+        // carry a delta starting at id 1 (above alpha), not 0.
+        assertMemoryLeak(() -> {
+            CatchUpHandler handler = new CatchUpHandler();
+            try (TestWebSocketServer server = new TestWebSocketServer(handler)) {
+                int port = server.getPort();
+                server.start();
+                Assert.assertTrue(server.awaitStart(5, TimeUnit.SECONDS));
+
+                try (Sender sender = Sender.fromConfig("ws::addr=localhost:" + port + ";")) {
+                    // Connection 1 registers alpha (id 0); the server ACKs and drops it.
+                    sender.table("t").symbol("s", "alpha").longColumn("v", 1L).atNow();
+                    sender.flush();
+                    waitFor(() -> handler.dictFor(1).size() >= 1, 5_000);
+                    waitFor(() -> handler.conn1Closed, 5_000);
+
+                    // Connection 2 (fresh) registers beta (id 1). With the baseline
+                    // preserved, beta ships as a delta ABOVE id 0 (deltaStart=1).
+                    sender.table("t").symbol("s", "beta").longColumn("v", 2L).atNow();
+                    sender.flush();
+                    waitFor(() -> handler.connectionsAccepted.get() >= 2
+                            && handler.dictFor(2).size() >= 2, 5_000);
+                }
+
+                Assert.assertTrue("connection 2 must re-register the dictionary via a catch-up first",
+                        handler.sawZeroTableFrameOnConn2);
+                Assert.assertTrue("post-reconnect data frame must ship a delta ABOVE the surviving "
+                                + "baseline (deltaStart >= 1); a reset baseline re-ships the whole "
+                                + "dictionary from deltaStart 0",
+                        handler.conn2SawDeltaAboveBaseline);
+            }
+        });
+    }
+
     @Test
     public void testFixedCapNearBoundarySymbolCatchesUpWithoutTerminal() throws Exception {
         // Regression (homogeneous single cap): a symbol whose length sits just below
@@ -332,6 +378,13 @@ private static class CatchUpHandler implements TestWebSocketServer.WebSocketServ
         // Set from the flags byte of the zero-table catch-up frame on connection 2:
         // the catch-up carries no rows and must defer its (empty) commit.
         volatile boolean catchUpDeferredOnConn2;
+        // Set when connection 2 receives a DATA frame (tableCount > 0) whose delta
+        // starts ABOVE id 0 (deltaStart >= 1). This can only happen if the producer's
+        // monotonic baseline SURVIVED the reconnect: a reset would re-ship the whole
+        // dictionary from deltaStart 0. Robust to replay -- a replayed pre-reconnect
+        // frame carries its original deltaStart 0, so only a genuinely-above-baseline
+        // post-reconnect frame trips it.
+        volatile boolean conn2SawDeltaAboveBaseline;
         volatile boolean sawZeroTableFrameOnConn2;
         private final List<List<String>> dictsByConn = new CopyOnWriteArrayList<>();
         private TestWebSocketServer.ClientHandler currentClient;
@@ -356,10 +409,24 @@ public synchronized void onBinaryMessage(TestWebSocketServer.ClientHandler clien
             int connNumber = dictsByConn.size();
             List<String> dict = dictsByConn.get(connNumber - 1);
             accumulate(data, dict);
-            if (connNumber == 2 && tableCount(data) == 0) {
-                sawZeroTableFrameOnConn2 = true;
-                // FLAG_DEFER_COMMIT is bit 0x01 of the flags byte (offset 5).
-                catchUpDeferredOnConn2 = (data[5] & 0x01) != 0;
+            if (connNumber == 2) {
+                if (tableCount(data) == 0) {
+                    sawZeroTableFrameOnConn2 = true;
+                    // FLAG_DEFER_COMMIT is bit 0x01 of the flags byte (offset 5).
+                    catchUpDeferredOnConn2 = (data[5] & 0x01) != 0;
+                } else if (data.length >= 12 && (data[5] & 0x08) != 0) {
+                    // A post-reconnect DATA frame carrying a delta section
+                    // (FLAG_DELTA_SYMBOL_DICT = 0x08). A deltaStart >= 1 means the
+                    // producer resumed the delta ABOVE the surviving baseline; a reset
+                    // baseline would re-ship from deltaStart 0. Checking any frame (not
+                    // just the first) keeps this robust to a replayed pre-reconnect
+                    // frame arriving ahead of the new one -- that replay carries its
+                    // original deltaStart 0 and does not trip the flag.
+                    int[] pos = {12};
+                    if (readVarint(data, pos) >= 1) {
+                        conn2SawDeltaAboveBaseline = true;
+                    }
+                }
             }
             try {
                 client.sendBinary(buildAck(nextSeq.getAndIncrement()));
diff --git a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/SelfSufficientFramesTest.java b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/SelfSufficientFramesTest.java
@@ -25,6 +25,7 @@
 package io.questdb.client.test.cutlass.qwp.client;
 
 import io.questdb.client.Sender;
+import io.questdb.client.cutlass.line.LineSenderException;
 import io.questdb.client.test.cutlass.qwp.websocket.TestWebSocketServer;
 import org.junit.Assert;
 import org.junit.Test;
@@ -273,6 +274,64 @@ public void testSplitBatchShipsDeltaOnFirstFrameOnly() throws Exception {
         });
     }
 
+    @Test
+    public void testOversizedTableSplitStrandsNothing() throws Exception {
+        // Regression: flushPendingRowsSplit publishes each table's frame one at a
+        // time (all but the last deferred, i.e. appended but uncommitted). If a LATER
+        // table's frame exceeds the cap, the split must not have already published an
+        // earlier table's frame -- otherwise that prefix strands on the ring, a later
+        // commit delivers it as a partial batch, and resetTableBuffersAfterFlush
+        // discards every source row, all while flush() reported failure to the
+        // caller. The pre-flight size pass makes the split all-or-nothing: an
+        // oversized table throws BEFORE any frame is published. Pre-fix, the "small"
+        // table's frame was published and committed on close, so the server saw it.
+        assertMemoryLeak(() -> {
+            CapturingHandler handler = new CapturingHandler();
+            try (TestWebSocketServer server = new TestWebSocketServer(handler)) {
+                server.setAdvertisedMaxBatchSize(200);
+                int port = server.getPort();
+                server.start();
+                Assert.assertTrue(server.awaitStart(5, TimeUnit.SECONDS));
+
+                // auto_flush_bytes=off lets "big" accumulate PAST the cap (byte-based
+                // auto-flush is otherwise clamped to 90% of the cap and would flush
+                // first); the row/interval limits are set high so nothing auto-flushes
+                // during the test. Each row stays under the per-row guard (< cap), but
+                // 12 rows make "big"'s single frame exceed the cap, which no split can
+                // shrink. "small" (added first) fits; "big" (added second) does not, so
+                // the split hits it AFTER publishing "small" pre-fix.
+                String pad = new String(new char[40]).replace('\0', 'x');
+                try (Sender sender = Sender.fromConfig("ws::addr=localhost:" + port
+                        + ";auto_flush_bytes=off;auto_flush_rows=1000000;auto_flush_interval=60000;")) {
+                    sender.table("small").symbol("s", "a").longColumn("v", 1L).atNow();
+                    for (int i = 0; i < 12; i++) {
+                        sender.table("big").stringColumn("p", pad).longColumn("v", (long) i).atNow();
+                    }
+                    try {
+                        sender.flush();
+                        Assert.fail("an oversized single-table split frame must throw");
+                    } catch (LineSenderException e) {
+                        Assert.assertTrue(e.getMessage(),
+                                e.getMessage().contains("too large for server batch cap"));
+                    }
+                    // close() drains the ring: pre-fix, the stranded "small" frame
+                    // would be sent (and committed) here.
+                }
+
+                // No DATA frame (tableCount > 0) may have reached the server: the
+                // oversized-table split published nothing. Pre-fix, "small" arrived.
+                long dataFrames = 0;
+                for (byte[] frame : handler.batches) {
+                    if (frame.length >= 8 && (((frame[6] & 0xFF) | ((frame[7] & 0xFF) << 8)) > 0)) {
+                        dataFrames++;
+                    }
+                }
+                Assert.assertEquals("an oversized-table split must publish NO data frame -- an "
+                                + "earlier table's frame must not strand on the ring", 0, dataFrames);
+            }
+        });
+    }
+
     private static int readVarint(byte[] buf, int offset) {
         // Simple unsigned varint decode — sufficient for small values.
         int result = 0;
