Cover three untested-critical rows; explain the two that resist

C22. The full-dict-fallback discard is gated on BOTH
recoveredMaxSymbolId >= size AND recoveredMaxSymbolDeltaStart == 0, and
the second conjunct had no failing test. Every candidate passed either
way: the torn-dict tests have frames covering from id 0, so a re-fold at
baseline 0 succeeds anyway, and the covered-prefix test fails the size
conjunct first. The new test builds the shape that discriminates -- a
side-file holding [a,b] beside frames registering c@2 and d@3, so the
frames out-reach the dictionary but are NOT self-sufficient. Delete the
conjunct and recovery discards the only source of ids 0..1, re-folds at
0, finds deltaStart 2 above a coverage of 0 and quarantines a perfectly
recoverable slot.

C23. start()'s catch of CatchUpSendException had no test: no other case
combines a seeded mirror with a failing client through start(). It
matters because a recovered sender ships its catch-up on the CALLER's
thread in SYNC/OFF startup, so without the catch a server that drops
mid-catch-up turns a transient outage into a build() failure -- an
Invariant B violation. Reverted, the new test reports CatchUpSend
"transient wire failure during catch-up" straight out of start().

C25. Nothing exercised a split failing at frame k>1; every other
failed-publish test fails at frame 1. The new test injects at the RING
rather than the cap -- both frames clear the pre-flight, but the second
does not fit a fresh sf_max_bytes segment -- so frame 1 publishes and
frame 2 throws. It is labelled a CHARACTERISATION test, not a mutation
guard, because it does not discriminate against reverting the pd.size()
persist key: frame 1 published successfully, so advanceSentMaxSymbolId
had already moved the baseline past the whole batch and the retry is an
early return under either key. That is exactly why the k>1 path is safe,
and it is what the test records.

C21 and C24 are NOT covered, and both are downgraded.

C21 -- the advanceSentMaxSymbolId/sealAndSwapBuffer ordering -- needs a
publish that fails ONCE and then succeeds. Every failure mode reachable
from the public API is permanent for a given batch: an oversized frame
stays oversized, and the rows are retained deliberately, so the retry
re-encodes the same bytes. Reaching it would need a fault-injection seam
at appendBlocking, which is more production test-surface than the row
justifies now that C25 covers the k>1 shape.

C24 -- trySendOne's re-anchor catch and its cap-gap discrimination -- is
unreachable, for the reason that retracted C13 in the previous commit:
both connection setup sites call tryRetireOrphanTail before any send, so
that arm is only reached once frames below the tail have been sent on
this connection, and the catch sits inside the nextWireSeq == 0 branch
that then cannot hold.

Full client suite: 2701 tests green.

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

Author: glasstiger

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

@@ -273,6 +273,90 @@ public void testSplitBatchShipsDeltaOnFirstFrameOnly() throws Exception {
         });
     }
 
+    /**
+     * A split that fails at frame k>1 must leave the symbol dictionary consistent for
+     * the retry.
+     * <p>
+     * flushPendingRowsSplit publishes frames one at a time and documents that a failure
+     * partway through leaves frames 1..k-1 on the ring as deferred, with the next flush
+     * re-emitting the whole batch -- at-least-once, absorbed by DEDUP or a durable-ack
+     * await. Nothing exercised the k&gt;1 shape at all: every other failed-publish test
+     * fails at frame 1.
+     * <p>
+     * This is a CHARACTERISATION test, not a mutation guard, and the reason is worth
+     * recording. It confirms the javadoc's claim that "the re-sent frames carry empty
+     * deltas and the write-ahead persist is a pd.size() no-op" -- but it does not
+     * discriminate against reverting that pd.size() key, because frame 1 published
+     * SUCCESSFULLY and advanceSentMaxSymbolId therefore already moved the baseline past
+     * the whole batch. The retry is an early return under either key. That is precisely
+     * why the k&gt;1 path is safe, and it is the fact the test pins.
+     * <p>
+     * The failure is injected at the RING, not the cap: both frames pass the pre-flight
+     * (each is under the advertised cap) but "big"'s frame does not fit a fresh
+     * sf_max_bytes segment, so it fails in sealAndSwapBuffer AFTER "a" was published.
+     */
+    @Test(timeout = 60_000L)
+    public void testMidSplitPublishFailureLeavesTheDictionaryIdempotent() throws Exception {
+        Path sfDir = temporaryFolder.newFolder("qwp-sf-midsplit").toPath();
+        assertMemoryLeak(() -> {
+            CapturingHandler handler = new CapturingHandler();
+            try (TestWebSocketServer server = new TestWebSocketServer(handler)) {
+                server.setAdvertisedMaxBatchSize(3_400);
+                int port = server.getPort();
+                server.start();
+                Assert.assertTrue(server.awaitStart(5, TimeUnit.SECONDS));
+
+                String pad = new String(new char[100]).replace('\0', 'x');
+                String cfg = "ws::addr=localhost:" + port
+                        + ";sf_dir=" + sfDir
+                        + ";sf_max_bytes=1024"
+                        + ";auto_flush_bytes=off;auto_flush_rows=1000000;auto_flush_interval=60000"
+                        + ";close_flush_timeout_millis=0;";
+                Sender sender = Sender.fromConfig(cfg);
+                try {
+                    // "a" first (small: fits a segment), "big" second (does not).
+                    sender.table("a").symbol("s", "alpha").longColumn("v", 1L).atNow();
+                    for (int i = 0; i < 28; i++) {
+                        sender.table("big").symbol("s", "bravo")
+                                .stringColumn("p", pad).longColumn("v", (long) i).atNow();
+                    }
+
+                    ObjList<String> afterFirst = null;
+                    for (int attempt = 0; attempt < 2; attempt++) {
+                        try {
+                            sender.flush();
+                            Assert.fail("the second split frame cannot fit a fresh segment");
+                        } catch (LineSenderException expected) {
+                            // Ring-level failure, not the cap pre-flight.
+                            Assert.assertFalse("must fail at the ring, not the cap pre-flight: "
+                                            + expected.getMessage(),
+                                    expected.getMessage().contains("too large for server batch cap"));
+                        }
+                        ObjList<String> persisted =
+                                ((QwpWebSocketSender) sender).getPersistedSymbolsForTest();
+                        Assert.assertNotNull(persisted);
+                        if (attempt == 0) {
+                            afterFirst = persisted;
+                        } else {
+                            Assert.assertEquals("a mid-split retry must not re-append the "
+                                            + "symbols the failed attempt already persisted",
+                                    afterFirst.size(), persisted.size());
+                        }
+                    }
+                    Assert.assertEquals("both symbols are persisted exactly once",
+                            2, afterFirst.size());
+                } finally {
+                    try {
+                        // close() re-flushes the retained batch, which is permanently
+                        // unflushable at this sf_max_bytes -- not what we assert.
+                        sender.close();
+                    } catch (LineSenderException ignored) {
+                    }
+                }
+            }
+        });
+    }
+
     @Test
     public void testOversizedTableSplitStrandsNothing() throws Exception {
         // Regression: flushPendingRowsSplit publishes each table's frame one at a

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

@@ -1001,6 +1001,43 @@ private static void invokeConnectLoop(CursorWebSocketSendLoop loop) {
         loop.connectLoopForTest(new LineSenderException("test reconnect"), "reconnect", 0L);
     }
 
+    /**
+     * start()'s catch of CatchUpSendException must absorb a transient catch-up send
+     * failure, not let it out of Sender construction.
+     * <p>
+     * A recovered sender re-registers its dictionary with a catch-up on the very first
+     * connect, and in SYNC/OFF startup that runs on the CALLER's thread inside start().
+     * Without the catch, a server that drops during the catch-up turns a transient outage
+     * into a build() failure -- an Invariant B violation, and the one thing
+     * store-and-forward exists to prevent. Delete the block and everything else stays
+     * green: no other test combines a seeded mirror with a failing client through start().
+     * <p>
+     * The failure must also stay non-terminal: the client is dropped so the I/O thread
+     * reconnects and re-sends the catch-up off the caller's thread.
+     */
+    @Test
+    public void testStartAbsorbsATransientCatchUpSendFailure() throws Exception {
+        TestUtils.assertMemoryLeak(() -> {
+            try (CursorSendEngine engine = newEngine()) {
+                CatchUpCapturingClient client = new CatchUpCapturingClient(0, true);
+                CursorWebSocketSendLoop loop = newForegroundLoop(engine, client);
+                try {
+                    // One mirror entry is enough to make setWireBaselineWithCatchUp ship a
+                    // catch-up, which the client then fails. loop.close() frees addr.
+                    long addr = Unsafe.malloc(2, MemoryTag.NATIVE_DEFAULT);
+                    long p = writeVarint(addr, 1);
+                    Unsafe.getUnsafe().putByte(p, (byte) 'a');
+                    loop.seedSentDictMirrorForTest(addr, 2, 1);
+
+                    loop.start(); // must NOT throw: build() would fail on a transient
+                    loop.checkError(); // and must not have latched a terminal
+                } finally {
+                    loop.close();
+                }
+            }
+        });
+    }
+
     private CursorWebSocketSendLoop newLoop(CursorSendEngine engine, WebSocketClient client) {
         return newLoop(engine, client, 0L);
     }

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

@@ -542,6 +542,56 @@ public void testGapResetInsideTheDeferredTailFailsCleanInsteadOfMisattributing()
         });
     }
 
+    /**
+     * A torn DELTA slot must KEEP its dictionary: the full-dict-fallback discard is
+     * gated on {@code recoveredMaxSymbolDeltaStart == 0L} and that conjunct is
+     * load-bearing in the negative direction.
+     * <p>
+     * Here the side-file holds [a,b] and the surviving frames register c@2 and d@3 --
+     * so the frames out-reach the dictionary (recoveredMaxSymbolId 3 >= size 2, the
+     * other conjunct) but are NOT self-sufficient. Drop the deltaStart conjunct and
+     * recovery discards the only source of ids 0..1, re-folds at baseline 0, finds
+     * deltaStart 2 above a coverage of 0, marks a gap and quarantines a slot that is
+     * perfectly recoverable.
+     * <p>
+     * Every existing candidate passes either way: the torn-dict tests have frames
+     * covering from id 0, and testRecoverySkipsEntriesAlreadyCoveredByPersistedPrefix
+     * fails the size conjunct first, so neither one is load-bearing there.
+     */
+    @Test
+    public void testTornDeltaSlotKeepsItsDictionaryInsteadOfBeingDiscarded() throws Exception {
+        TestUtils.assertMemoryLeak(() -> {
+            try (CursorSendEngine engine = newEngine()) {
+                appendDeltaSymbolFrame(engine, 2, 'c');
+                appendDeltaSymbolFrame(engine, 3, 'd');
+            }
+            try (PersistedSymbolDict pd = PersistedSymbolDict.openClean(tmpDir)) {
+                assertNotNull(pd);
+                pd.appendSymbol("a");
+                pd.appendSymbol("b");
+            }
+
+            try (CursorSendEngine engine = newEngine()) {
+                assertNotNull("a torn DELTA slot must keep its dictionary -- its frames "
+                                + "cannot rebuild the ids below their own delta start",
+                        engine.getPersistedSymbolDict());
+                assertEquals("scaffolding: the frames must out-reach the dictionary, so the "
+                                + "OTHER discard conjunct is satisfied and this test really "
+                                + "pins the deltaStart one",
+                        3L, engine.recoveredMaxSymbolId());
+                assertEquals(2, engine.getPersistedSymbolDict().size());
+
+                GlobalSymbolDictionary recovered = new GlobalSymbolDictionary();
+                assertEquals("recovery must cover ids 0..3, not fail clean",
+                        4L, engine.addRecoveredSymbolsTo(2, recovered));
+                assertEquals("only the suffix above the persisted prefix is added",
+                        2, recovered.size());
+                assertEquals("c", recovered.getSymbol(0));
+                assertEquals("d", recovered.getSymbol(1));
+            }
+        });
+    }
+
     @Test
     public void testSelfSufficientFrameRepairsAckedRecoveryGap() throws Exception {
         // fsn 0 models the tail of an old delta epoch whose registering frames