Free the recovery mirror on a ctor seed throw

The CursorWebSocketSendLoop constructor mallocs the sent-dictionary
mirror from the persisted dictionary's prefix, then extends it from the
surviving frames' deltas. If that second step throws -- a native realloc
OOM, or the MAX_SENT_DICT_BYTES ceiling -- the throw leaves the loop
unpublished, so neither ensureConnected's catch nor BackgroundDrainer's
finally can close() it, and the already-malloc'd mirror leaks. Free it on
any throw, mirroring the loopNeverRan free in close()/ioLoop's exit.

Also close two test-coverage gaps the review flagged:

- The send loop's torn-dict guard (deltaStart > sentDictCount in
  trySendOne) is the drainer path's sole defense against replaying a
  frame whose symbols were trimmed away; the foreground path is
  quarantined earlier, so this fire direction was unexercised. Cover it
  at the send-loop level against a stub client (the drainer integration
  path cannot connect on darwin), asserting zero frames ship and the
  terminal latches.

- quarantineTornSlot must fail loudly rather than drop bytes when it
  cannot set a slot aside. Saturate every quarantine candidate and
  assert build() throws and preserves the slot on disk.

A TestOnly seam (forceMirrorSeedFailureForTest) makes the ctor seed
throw deterministically; the unreplayable-slot setup is now a shared
helper.

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

@@ -197,6 +197,14 @@ public final class CursorWebSocketSendLoop implements QuietCloseable {
      * Throttle "reconnect attempt N failed" WARN logs to one per 5 s.
      */
     private static final long RECONNECT_LOG_THROTTLE_NANOS = 5_000_000_000L;
+    // Test seam: when true, appendSymbolToMirror throws instead of appending,
+    // simulating the native realloc OOM (or MAX_SENT_DICT_BYTES ceiling) that
+    // ensureSentDictCapacity can raise while the constructor seeds the recovery
+    // mirror. Lets a test prove the constructor frees the already-malloc'd mirror on
+    // such a throw rather than leaking it. Production never sets it; volatile only so
+    // a test thread's write is visible to the loop under test.
+    @TestOnly
+    static volatile boolean forceMirrorSeedFailureForTest;
     // Pre-converted to nanos for the comparison in sendDurableAckKeepaliveIfDue.
     // Zero or negative disables the keepalive entirely.
     private final long durableAckKeepaliveIntervalNanos;
@@ -717,11 +725,30 @@ public CursorWebSocketSendLoop(WebSocketClient client, CursorSendEngine engine,
         // nothing and cost a catch-up frame on every connection -- full-dict slots must stay
         // catch-up-free.
         if (engine.recoveredMaxSymbolDeltaStart() > 0L) {
-            ObjList<String> fromFrames = new ObjList<>();
-            if (engine.collectReplaySymbolsAbove(sentDictCount, fromFrames) >= 0) {
-                for (int i = 0, n = fromFrames.size(); i < n; i++) {
-                    appendSymbolToMirror(fromFrames.getQuick(i));
+            // The prefix seed above may already have malloc'd the mirror, and
+            // appendSymbolToMirror -> ensureSentDictCapacity can still throw here (a
+            // native realloc OOM, or the MAX_SENT_DICT_BYTES ceiling). Such a throw
+            // propagates out of the constructor, so the half-built loop is never
+            // assigned to a reference -- neither ensureConnected's catch nor
+            // BackgroundDrainer's finally can close() it -- and the mirror would leak.
+            // Free it on any throw so the constructor leaves nothing behind, mirroring
+            // the loopNeverRan free in close() / ioLoop's exit.
+            try {
+                ObjList<String> fromFrames = new ObjList<>();
+                if (engine.collectReplaySymbolsAbove(sentDictCount, fromFrames) >= 0) {
+                    for (int i = 0, n = fromFrames.size(); i < n; i++) {
+                        appendSymbolToMirror(fromFrames.getQuick(i));
+                    }
                 }
+            } catch (Throwable t) {
+                if (sentDictBytesAddr != 0) {
+                    Unsafe.free(sentDictBytesAddr, sentDictBytesCapacity, MemoryTag.NATIVE_DEFAULT);
+                    sentDictBytesAddr = 0;
+                    sentDictBytesCapacity = 0;
+                    sentDictBytesLen = 0;
+                    sentDictCount = 0;
+                }
+                throw t;
             }
         }
         this.fsnAtZero = fsnAtZero;
@@ -2299,6 +2326,10 @@ private void accumulateSentDict(long payloadAddr, int payloadLen, int deltaStart
      * construction; {@link #accumulateSentDict} extends it from live frames thereafter.
      */
     private void appendSymbolToMirror(CharSequence symbol) {
+        if (forceMirrorSeedFailureForTest) {
+            // Simulate a native realloc OOM on the seed path (see the field).
+            throw new LineSenderException("simulated mirror seed allocation failure (test only)");
+        }
         int utf8Len = Utf8s.utf8Bytes(symbol);
         int wireLen = NativeBufferWriter.varintSize(utf8Len) + utf8Len;
         ensureSentDictCapacity((long) sentDictBytesLen + wireLen);

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

@@ -223,32 +223,7 @@ public void testTrimmedRegisteringFramesAreUnreplayableAndTheSlotIsSetAside() th
         // retained on close, so every retry re-recovered the same slot and threw again, and the
         // application could not construct a Sender at all -- it could not even buffer new rows.
         assertMemoryLeak(() -> {
-            try (TestWebSocketServer silent = new TestWebSocketServer(new SilentHandler())) {
-                int port = silent.getPort();
-                silent.start();
-                Assert.assertTrue(silent.awaitStart(5, TimeUnit.SECONDS));
-                // Small segments so the frames roll into several files and the earliest ones
-                // can be trimmed away independently.
-                String cfg = "ws::addr=localhost:" + port + ";sf_dir=" + sfDir
-                        + ";sf_max_bytes=256;close_flush_timeout_millis=0;";
-                try (Sender s1 = Sender.fromConfig(cfg)) {
-                    for (int i = 0; i < 12; i++) {
-                        s1.table("m").symbol("s", "sym-" + i).longColumn("v", i).atNow();
-                        s1.flush();
-                    }
-                }
-            }
-
-            java.nio.file.Path slot = Paths.get(sfDir, "default");
-            Assert.assertTrue("the frames must have rolled into more than one segment",
-                    countSegmentFiles(slot) > 1);
-            // TRIM the segment that holds the earliest frames -- munmap + unlink is exactly what
-            // SegmentManager does once they are acked. Their symbols are now gone from disk.
-            java.nio.file.Files.delete(slot.resolve("sf-initial.sfa"));
-            // ...and the dictionary is torn away as well, so nothing holds them at all.
-            java.nio.file.Path dict = slot.resolve(".symbol-dict");
-            java.nio.file.Files.write(dict,
-                    Arrays.copyOf(java.nio.file.Files.readAllBytes(dict), 8));
+            writeAndTearUnreplayableSlot();
 
             DictReconstructingHandler handler = new DictReconstructingHandler();
             try (TestWebSocketServer good = new TestWebSocketServer(handler)) {
@@ -287,6 +262,82 @@ private static int countSegmentFiles(java.nio.file.Path dir) {
         return n;
     }
 
+    @Test
+    public void testQuarantineFailsLoudlyWhenAllSlotNamesSaturated() throws Exception {
+        // M2 regression: when a recovered slot is genuinely unreplayable, build() sets
+        // it aside (quarantineTornSlot) and starts the producer on a fresh slot. But if
+        // it cannot free the slot name -- here every default.unreplayable-<i> candidate
+        // up to MAX_QUARANTINE_SLOT_ATTEMPTS (64) already exists -- it MUST fail LOUDLY:
+        // throw and leave the slot's bytes on disk for a manual resend, never silently
+        // drop data. Only the happy rename path was covered before.
+        assertMemoryLeak(() -> {
+            writeAndTearUnreplayableSlot();
+            // Saturate every quarantine candidate so quarantineTornSlot's rename loop
+            // finds no free name.
+            for (int i = 0; i < 64; i++) {
+                java.nio.file.Files.createDirectories(Paths.get(sfDir, "default.unreplayable-" + i));
+            }
+
+            java.nio.file.Path slot = Paths.get(sfDir, "default");
+            try (TestWebSocketServer good = new TestWebSocketServer(new DictReconstructingHandler())) {
+                int port = good.getPort();
+                good.start();
+                Assert.assertTrue(good.awaitStart(5, TimeUnit.SECONDS));
+                String cfg = "ws::addr=localhost:" + port + ";sf_dir=" + sfDir + ";";
+                Sender s = null;
+                try {
+                    s = Sender.fromConfig(cfg);
+                    Assert.fail("build() must throw when the unreplayable slot cannot be set aside");
+                } catch (LineSenderException expected) {
+                    Assert.assertTrue("unexpected message: " + expected.getMessage(),
+                            expected.getMessage().contains("too many quarantined slots already under")
+                                    && expected.getMessage().contains("moved or removed by hand"));
+                } finally {
+                    if (s != null) {
+                        s.close();
+                    }
+                }
+            }
+            // The unreplayable slot's bytes must survive on disk for a manual resend --
+            // the guard fails loudly rather than dropping data.
+            Assert.assertTrue("the slot dir must be preserved", java.nio.file.Files.exists(slot));
+            Assert.assertTrue("the slot's segment data must be preserved",
+                    countSegmentFiles(slot) >= 1);
+        });
+    }
+
+    // Writes 12 delta frames (each introducing a new symbol) into the default slot
+    // across several small segments, then makes the slot GENUINELY unreplayable: trims
+    // the segment holding the earliest ids (munmap + unlink, exactly what SegmentManager
+    // does once they are acked) and tears the .symbol-dict down to its header, so the
+    // surviving frames' deltas start above ids nothing on disk still holds. Recovering
+    // such a slot throws UnreplayableSlotException.
+    private void writeAndTearUnreplayableSlot() throws Exception {
+        try (TestWebSocketServer silent = new TestWebSocketServer(new SilentHandler())) {
+            int port = silent.getPort();
+            silent.start();
+            Assert.assertTrue(silent.awaitStart(5, TimeUnit.SECONDS));
+            // Small segments so the frames roll into several files and the earliest ones
+            // can be trimmed away independently.
+            String cfg = "ws::addr=localhost:" + port + ";sf_dir=" + sfDir
+                    + ";sf_max_bytes=256;close_flush_timeout_millis=0;";
+            try (Sender s1 = Sender.fromConfig(cfg)) {
+                for (int i = 0; i < 12; i++) {
+                    s1.table("m").symbol("s", "sym-" + i).longColumn("v", i).atNow();
+                    s1.flush();
+                }
+            }
+        }
+        java.nio.file.Path slot = Paths.get(sfDir, "default");
+        Assert.assertTrue("the frames must have rolled into more than one segment",
+                countSegmentFiles(slot) > 1);
+        // TRIM the segment that holds the earliest frames.
+        java.nio.file.Files.delete(slot.resolve("sf-initial.sfa"));
+        // ...and tear the dictionary away as well, so nothing holds those ids at all.
+        java.nio.file.Path dict = slot.resolve(".symbol-dict");
+        java.nio.file.Files.write(dict, Arrays.copyOf(java.nio.file.Files.readAllBytes(dict), 8));
+    }
+
     @Test
     public void testUnopenableDictSeedsTheProducerAboveTheRecoveredIds() throws Exception {
         // The producer must resume ABOVE the ids the recovered frames already define, even when

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

@@ -25,6 +25,7 @@
 package io.questdb.client.test.cutlass.qwp.client.sf.cursor;
 
 import io.questdb.client.Sender;
+import io.questdb.client.cutlass.line.LineSenderException;
 import io.questdb.client.cutlass.qwp.client.sf.cursor.CursorSendEngine;
 import io.questdb.client.cutlass.qwp.client.sf.cursor.CursorWebSocketSendLoop;
 import io.questdb.client.cutlass.qwp.client.sf.cursor.PersistedSymbolDict;
@@ -156,6 +157,63 @@ public void testRecycledLoopReSeedsMirrorFromPersistedDict() throws Exception {
         }
     }
 
+    @Test
+    public void testCtorFreesSeededMirrorWhenFrameSeedThrows() throws Exception {
+        // C1 regression: the constructor seeds the recovery mirror in TWO steps -- a
+        // malloc from the persisted dictionary's intact prefix, then an extension from
+        // the surviving frames' own deltas (appendSymbolToMirror). If that second step
+        // throws (a native realloc OOM, or the MAX_SENT_DICT_BYTES ceiling), the throw
+        // leaves the constructor with the object unpublished, so neither
+        // ensureConnected's catch nor BackgroundDrainer's finally can ever close() it --
+        // and the already-malloc'd prefix mirror leaks. The constructor must free it on
+        // the throw. Pre-fix, the mirror leaks here and assertMemoryLeak fails.
+        Path sfDir = Files.createTempDirectory("qwp-mirror-ctor-throw");
+        try {
+            // A torn-dict SUBSET: three delta frames a@0,b@1,c@2 survive on disk, but the
+            // .symbol-dict is rewritten to hold only [a,b] (a host-crash tail tear). On
+            // recovery pd.size()==2 seeds (mallocs) the mirror, then the frame-seed
+            // rebuilds c@2 from the surviving frame -- the append the fault interrupts.
+            populateThreeFrameSlot(sfDir);
+            Path slot = sfDir.resolve("default");
+            try (PersistedSymbolDict torn = PersistedSymbolDict.openClean(slot.toString())) {
+                Assert.assertNotNull(torn);
+                torn.appendSymbol("a");
+                torn.appendSymbol("b");
+                Assert.assertEquals(2, torn.size());
+            }
+
+            assertMemoryLeak(() -> {
+                try (CursorSendEngine engine = new CursorSendEngine(slot.toString(), 4096)) {
+                    PersistedSymbolDict pd = engine.getPersistedSymbolDict();
+                    Assert.assertNotNull("recovery must open the torn subset dict", pd);
+                    Assert.assertEquals("prefix seed must malloc a 2-entry mirror", 2, pd.size());
+                    Assert.assertTrue("the frame-seed path must run (frames out-reach the dict)",
+                            engine.recoveredMaxSymbolDeltaStart() > 0L);
+
+                    setMirrorSeedFault(true);
+                    try {
+                        new CursorWebSocketSendLoop(
+                                null, engine, 0, 1_000_000L,
+                                () -> {
+                                    throw new IOException("no reconnect in this test");
+                                },
+                                0, 0, 1);
+                        Assert.fail("ctor must propagate the injected mirror-seed failure");
+                    } catch (LineSenderException expected) {
+                        Assert.assertTrue("unexpected message: " + expected.getMessage(),
+                                expected.getMessage().contains("simulated mirror seed allocation failure"));
+                    } finally {
+                        setMirrorSeedFault(false);
+                    }
+                    // The outer assertMemoryLeak proves the prefix-seeded mirror the ctor
+                    // malloc'd was freed on the throw -- pre-fix it leaks here.
+                }
+            });
+        } finally {
+            rmDir(sfDir);
+        }
+    }
+
     // Constructs a recovery send loop but does NOT start it: the ctor seeds the
     // catch-up mirror synchronously, which is all these tests observe. The
     // reconnect factory is never invoked.
@@ -189,12 +247,43 @@ private static void populateRecoverableSlot(Path sfDir) throws Exception {
         }
     }
 
+    // Three delta frames a@0, b@1, c@2, nothing acked, so all three survive and
+    // replay from frame 0. Paired with a dictionary truncated to [a,b], this is a
+    // torn-dict SUBSET whose recovery drives the constructor's frame-seed path.
+    private static void populateThreeFrameSlot(Path sfDir) throws Exception {
+        try (TestWebSocketServer silent = new TestWebSocketServer(new SilentHandler())) {
+            int port = silent.getPort();
+            silent.start();
+            Assert.assertTrue(silent.awaitStart(5, TimeUnit.SECONDS));
+            String cfg = "ws::addr=localhost:" + port
+                    + ";sf_dir=" + sfDir
+                    + ";close_flush_timeout_millis=0;";
+            try (Sender s = Sender.fromConfig(cfg)) {
+                s.table("m").symbol("s", "a").longColumn("v", 0).atNow();
+                s.flush();
+                s.table("m").symbol("s", "b").longColumn("v", 1).atNow();
+                s.flush();
+                s.table("m").symbol("s", "c").longColumn("v", 2).atNow();
+                s.flush();
+            }
+        }
+    }
+
     private static int readInt(CursorWebSocketSendLoop loop, String name) throws Exception {
         Field f = CursorWebSocketSendLoop.class.getDeclaredField(name);
         f.setAccessible(true);
         return f.getInt(loop);
     }
 
+    // Toggles the loop's @TestOnly mirror-seed fault flag. Reflection because the
+    // flag is package-private in the production package (this test is in a sibling
+    // test package), the same non-reflective-path-unavailable reason readInt uses.
+    private static void setMirrorSeedFault(boolean value) throws Exception {
+        Field f = CursorWebSocketSendLoop.class.getDeclaredField("forceMirrorSeedFailureForTest");
+        f.setAccessible(true);
+        f.setBoolean(null, value);
+    }
+
     private static void rmDir(Path dir) throws IOException {
         TestUtils.removeTmpDirRec(dir == null ? null : dir.toString());
     }