Don't quarantine a fully-acked torn-dict slot

seedGlobalDictionaryFromPersisted walked collectReplaySymbolsAbove from
FSN 0 and threw UnreplayableSlotException on the first symbol-dictionary
gap regardless of ack state. For a recovered slot whose committed frames
were all acked -- so nothing was left to replay -- a gap in that already
delivered data still threw. (A host crash tore the unsynced side-file,
or it could not be opened, while the low segments that registered the
missing ids were trimmed away.) build() then set the slot aside as
<sender>.unreplayable-<n>, its fully-drained close deleted the delivered
bytes the quarantine promises to preserve, and it logged a false "resend
required" for data the server had already acknowledged.

That gap only matters for frames that will REPLAY. When every committed
frame has been acked (ackedFsn >= recoveredCommitBoundaryFsn), the seed
now suppresses the throw: it seeds the intact prefix only, leaving the
frame-derived symbols unused exactly as the send loop's mirror does on a
-1, so the producer baseline and the mirror's sentDictCount still agree
by construction, and resumes on the same slot. A genuine gap in UNACKED
committed frames still throws, so the real-unreplayable quarantine path
is unchanged.

Add testFullyAckedTornSlotResumesInPlaceWithoutQuarantine, the acked
counterpart to the existing unacked quarantine test: the same on-disk
tear, but with the ack watermark stamped at the last committed frame,
asserting no set-aside copy exists and the sender resumes in place. The
test fails without the production change.

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

@@ -3933,6 +3933,33 @@ private void seedGlobalDictionaryFromPersisted(PersistedSymbolDict pd) {
         ObjList<String> fromFrames = new ObjList<>();
         long coverage = cursorEngine.collectReplaySymbolsAbove(baseline, fromFrames);
         if (coverage < 0) {
+            // A gap: the surviving frames reference ids below their own delta start,
+            // introduced by frames since acked and trimmed away, and the persisted
+            // dictionary no longer holds them (a host crash tore its unsynced tail, or it
+            // could not be opened). That gap only matters for frames that will REPLAY.
+            // When every recovered committed frame is already acked
+            // (ackedFsn >= recoveredCommitBoundaryFsn), NOTHING replays: the gap is in
+            // data the server already has, and the retired orphan-deferred tail above the
+            // commit boundary is never transmitted. Throwing here would raise a false
+            // "resend required" for delivered data AND -- because such a slot is fully
+            // drained -- let build()'s connect-path close unlink the (already-delivered)
+            // bytes the quarantine claims to preserve. So DON'T throw: seed the intact
+            // prefix only, leaving fromFrames unused exactly as the send loop's mirror
+            // does on a -1 (it adds nothing), so the producer baseline and the mirror's
+            // sentDictCount still agree by construction. The producer resumes above the
+            // prefix and the fully-drained slot is cleaned up on close.
+            long ackedFsn = cursorEngine.ackedFsn();
+            long commitBoundaryFsn = cursorEngine.recoveredCommitBoundaryFsn();
+            if (ackedFsn >= commitBoundaryFsn) {
+                sentMaxSymbolId = globalSymbolDictionary.size() - 1;
+                LOG.info("recovered store-and-forward slot has a torn/incomplete symbol dictionary, "
+                                + "but every committed frame was already acked so nothing needs replaying; "
+                                + "resuming on the intact prefix without quarantine and without data loss "
+                                + "[recoveredPrefixSize={}, ackedFsn={}, commitBoundaryFsn={}]",
+                        baseline, ackedFsn, commitBoundaryFsn);
+                return;
+            }
+            // Genuine loss: unacked committed frames reference ids nothing still holds.
             // Typed, because Sender.build() sets such a slot aside instead of failing: this
             // is the point at which every source of truth has been tried and none of them
             // holds the missing ids. See UnreplayableSlotException.

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

@@ -249,6 +249,62 @@ public void testTrimmedRegisteringFramesAreUnreplayableAndTheSlotIsSetAside() th
         });
     }
 
+    @Test
+    public void testFullyAckedTornSlotResumesInPlaceWithoutQuarantine() throws Exception {
+        // M1 regression -- the ACKED counterpart to
+        // testTrimmedRegisteringFramesAreUnreplayableAndTheSlotIsSetAside. The on-disk
+        // tear is IDENTICAL (earliest segment trimmed, .symbol-dict torn to its header,
+        // so the surviving frames' deltas start above ids nothing on disk still holds),
+        // but here every committed frame was already ACKED before the crash. Nothing is
+        // left to replay, so the "gap" is entirely in data the server already has.
+        //
+        // seedGlobalDictionaryFromPersisted must therefore NOT raise
+        // UnreplayableSlotException. Quarantining a fully-delivered slot would fire a
+        // false "resend required" alarm AND -- because such a slot is fully drained --
+        // let build()'s connect-path close unlink the (already-delivered) bytes the
+        // quarantine claims to preserve. The slot must resume IN PLACE.
+        //
+        // Before the fix the gap detector ignored ack state and threw, so this slot was
+        // set aside as default.unreplayable-0 with a "resend required" error for data the
+        // server had already acknowledged.
+        assertMemoryLeak(() -> {
+            writeAndTearUnreplayableSlot();
+            // Mark every committed frame acked. writeAndTearUnreplayableSlot writes 12
+            // frames (FSNs 0..11), so stamping the watermark at 11 makes
+            // ackedFsn == recoveredCommitBoundaryFsn: a torn dictionary with nothing left
+            // to replay. (Not higher than 11 -- that would make the resuming producer's
+            // next frame look pre-acked.)
+            java.nio.file.Path slot = Paths.get(sfDir, "default");
+            writeAckWatermark(slot.resolve(".ack-watermark"), 11);
+
+            DictReconstructingHandler handler = new DictReconstructingHandler();
+            try (TestWebSocketServer good = new TestWebSocketServer(handler)) {
+                int port = good.getPort();
+                good.start();
+                Assert.assertTrue(good.awaitStart(5, TimeUnit.SECONDS));
+                String cfg = "ws::addr=localhost:" + port + ";sf_dir=" + sfDir + ";";
+                // build() must succeed WITHOUT setting the slot aside, and the producer
+                // must keep working on the SAME slot.
+                try (Sender s2 = Sender.fromConfig(cfg)) {
+                    s2.table("m").symbol("s", "after-recovery").longColumn("v", 99).atNow();
+                    s2.flush();
+                    long deadline = System.currentTimeMillis() + 10_000;
+                    while (System.currentTimeMillis() < deadline && handler.maxDictSize() < 1) {
+                        Thread.sleep(20);
+                    }
+                }
+            }
+            // The fully-acked slot was NOT quarantined: no set-aside copy exists (the
+            // inverse of assertUnreplayableSlotSetAside), and the sender resumed on the
+            // original slot.
+            Assert.assertFalse("a fully-acked torn slot must NOT be quarantined -- its data "
+                            + "was already delivered, so there is nothing to resend",
+                    java.nio.file.Files.isDirectory(Paths.get(sfDir, "default.unreplayable-0")));
+            Assert.assertTrue("the sender must resume on the original slot",
+                    java.nio.file.Files.isDirectory(slot));
+        });
+    }
+
     private static int countSegmentFiles(java.nio.file.Path dir) {
         java.io.File[] files = dir.toFile().listFiles();
         int n = 0;