Fix orphan-drain losing symbol-dict mirror seed

The send loop seeded its reconnect catch-up mirror by calling
PersistedSymbolDict.takeLoadedEntries(), a one-shot ownership transfer
that zeroed the dictionary's loaded-entries buffer. The foreground
sender builds a single loop, so that was safe. But BackgroundDrainer
builds a fresh send loop per wire session against the SAME, persistent
engine when a durable-ack capability gap forces a mid-drain recycle.
The second loop then found an empty loaded-entries buffer, seeded an
empty mirror (sentDictCount = 0), sent no reconnect catch-up, and the
first replayed delta frame (deltaStart > 0) tripped the torn-dict guard
-- falsely quarantining a healthy slot with a bogus "resend required"
terminal and defeating the capability-gap settle-retry for delta slots.

The send loop now COPIES the loaded entries into its own mirror and
leaves the dictionary owning its buffer for the engine's lifetime, so
every recycled loop re-seeds. PersistedSymbolDict.close() frees the
dictionary's copy; each loop frees its own copy on exit. Peak footprint
is unchanged (the drainer closes loop N before building loop N+1).

Removes the now-unused takeLoadedEntries() and loadedEntriesTaken, and
adds a regression test that builds two loops against one recovered
engine and asserts the second re-seeds (pre-fix its mirror was empty).

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

@@ -543,15 +543,21 @@ public CursorWebSocketSendLoop(WebSocketClient client, CursorSendEngine engine,
             if (pd != null && pd.size() > 0) {
                 int len = pd.loadedEntriesLen();
                 if (len > 0) {
-                    // Adopt the persisted dictionary's loaded-entries buffer as the
-                    // mirror's initial backing instead of copying it and leaving the
-                    // dictionary to retain a second copy for the engine's lifetime.
-                    // The producer's readLoadedSymbols() -- the only other consumer --
-                    // runs earlier in setCursorEngine; the drainer path has no
-                    // producer consumer. takeLoadedEntries() transfers ownership, so
-                    // this loop's mirror lifecycle frees it (not pd.close()). The
-                    // buffer's allocated size equals loadedEntriesLen.
-                    sentDictBytesAddr = pd.takeLoadedEntries();
+                    // COPY the persisted dictionary's loaded-entries buffer into this
+                    // loop's own mirror rather than taking ownership of it. The engine
+                    // (and its PersistedSymbolDict) OUTLIVES this loop on the orphan
+                    // drainer path: BackgroundDrainer builds a fresh send loop per wire
+                    // session against the same engine on a durable-ack capability-gap
+                    // recycle. A one-shot ownership transfer would leave every loop
+                    // after the first with an EMPTY mirror -- it would then send no
+                    // reconnect catch-up, and the first replayed delta frame
+                    // (deltaStart > 0) would trip the torn-dict guard, falsely
+                    // quarantining a healthy slot. Copying keeps the dictionary's
+                    // loaded entries intact for the engine's lifetime so every
+                    // recycled loop re-seeds; pd.close() (at engine close) frees the
+                    // dictionary's copy, this loop frees its own copy on exit.
+                    sentDictBytesAddr = Unsafe.malloc(len, MemoryTag.NATIVE_DEFAULT);
+                    Unsafe.getUnsafe().copyMemory(pd.loadedEntriesAddr(), sentDictBytesAddr, len);
                     sentDictBytesCapacity = len;
                     sentDictBytesLen = len;
                     // Set the count only alongside the bytes so sentDictCount can

core/src/main/java/io/questdb/client/cutlass/qwp/client/sf/cursor/PersistedSymbolDict.java

@@ -95,15 +95,14 @@ public final class PersistedSymbolDict implements QuietCloseable {
     private boolean closed;
     // In-memory copy of the entry region [len][utf8]... exactly as on disk,
     // populated only when open() recovered existing entries (recovery /
-    // orphan-drain). Zero/empty for a freshly created file. Consumed once to seed
-    // the producer's id map (readLoadedSymbols) and then ADOPTED by the send loop's
-    // catch-up mirror (takeLoadedEntries), which takes ownership so this file no
-    // longer retains a second copy of the dictionary for the engine's lifetime.
+    // orphan-drain). Zero/empty for a freshly created file. READ (not consumed) to
+    // seed the producer's id map (readLoadedSymbols) and to seed the send loop's
+    // catch-up mirror, which COPIES it. This file retains ownership for the engine's
+    // lifetime -- the orphan drainer builds a fresh send loop per wire session
+    // against the same engine, and each must re-seed its mirror -- and frees this
+    // buffer in close().
     private long loadedEntriesAddr;
     private int loadedEntriesLen;
-    // True once takeLoadedEntries() handed loadedEntriesAddr to the send-loop
-    // mirror; guards a stale readLoadedSymbols() call (which must run first).
-    private boolean loadedEntriesTaken;
     private long scratchAddr;
     private int scratchCap;
     private int size;
@@ -282,7 +281,6 @@ public int loadedEntriesLen() {
      * recovered.
      */
     public ObjList<String> readLoadedSymbols() {
-        assert !loadedEntriesTaken : "readLoadedSymbols() must run before takeLoadedEntries()";
         ObjList<String> out = new ObjList<>(Math.max(size, 1));
         long p = loadedEntriesAddr;
         long limit = p + loadedEntriesLen;
@@ -313,25 +311,6 @@ public int size() {
         return size;
     }
 
-    /**
-     * Transfers ownership of the loaded-entries buffer to the caller and returns
-     * its base address (0 when nothing was recovered). The send loop adopts it as
-     * the initial backing of its catch-up mirror rather than copying it, so this
-     * file stops retaining a second copy of the dictionary for the engine's
-     * lifetime. After this call {@link #close()} no longer frees the buffer -- the
-     * caller owns it. The producer's {@link #readLoadedSymbols()} is the only other
-     * consumer and MUST run first: {@code setCursorEngine} seeds the producer
-     * before the send loop is constructed, and the drainer path has no producer
-     * consumer. The buffer was allocated with {@link MemoryTag#NATIVE_DEFAULT}.
-     */
-    public long takeLoadedEntries() {
-        long addr = loadedEntriesAddr;
-        loadedEntriesAddr = 0L;
-        loadedEntriesLen = 0;
-        loadedEntriesTaken = true;
-        return addr;
-    }
-
     private static PersistedSymbolDict openExisting(String filePath, long fileLen) {
         int fd = Files.openRW(filePath);
         if (fd < 0) {

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

@@ -33,6 +33,7 @@
 import org.junit.Test;
 
 import java.io.IOException;
+import java.lang.reflect.Field;
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.util.Comparator;
@@ -96,6 +97,69 @@ public void testSeededMirrorFreedWhenLoopClosedWithoutStart() throws Exception {
         }
     }
 
+    @Test
+    public void testRecycledLoopReSeedsMirrorFromPersistedDict() throws Exception {
+        // C1 regression: the orphan drainer (BackgroundDrainer) builds a NEW
+        // CursorWebSocketSendLoop per wire session against the SAME, persistent
+        // engine when a durable-ack capability gap forces a mid-drain recycle. The
+        // recovery mirror seed must survive that recycle. If the first loop CONSUMED
+        // the persisted dictionary's loaded entries (a one-shot ownership transfer),
+        // the second loop seeds an EMPTY mirror (sentDictCount = 0), sends no
+        // reconnect catch-up, and the first replayed delta frame (deltaStart > 0)
+        // trips the torn-dict guard -- falsely quarantining a healthy slot with a
+        // bogus "resend required" terminal. Copying the entries (leaving the
+        // dictionary intact for the engine's lifetime) lets every recycled loop
+        // re-seed. Pre-fix, loop2's sentDictCount is 0 and this assertion fails.
+        Path sfDir = Files.createTempDirectory("qwp-mirror-reseed");
+        try {
+            populateRecoverableSlot(sfDir);
+            Path slot = sfDir.resolve("default");
+            assertMemoryLeak(() -> {
+                try (CursorSendEngine engine = new CursorSendEngine(slot.toString(), 4096)) {
+                    PersistedSymbolDict pd = engine.getPersistedSymbolDict();
+                    Assert.assertNotNull(pd);
+                    int dictSize = pd.size();
+                    Assert.assertTrue("recovery must load a non-empty dictionary", dictSize > 0);
+
+                    // Session 1 seeds its mirror from the persisted dictionary.
+                    CursorWebSocketSendLoop loop1 = newRecoveryLoop(engine);
+                    try {
+                        Assert.assertEquals("session-1 mirror must seed from the persisted dict",
+                                dictSize, readInt(loop1, "sentDictCount"));
+                    } finally {
+                        loop1.close();
+                    }
+
+                    // Session 2 against the SAME engine (the drainer recycle): the
+                    // seed must NOT have been consumed -- the mirror must re-seed to
+                    // the full dictionary so the reconnect catch-up is complete.
+                    CursorWebSocketSendLoop loop2 = newRecoveryLoop(engine);
+                    try {
+                        Assert.assertEquals("recycled session-2 mirror must re-seed from the "
+                                        + "persisted dict (pre-fix it was 0)",
+                                dictSize, readInt(loop2, "sentDictCount"));
+                    } finally {
+                        loop2.close();
+                    }
+                }
+            });
+        } 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.
+    private static CursorWebSocketSendLoop newRecoveryLoop(CursorSendEngine engine) {
+        return new CursorWebSocketSendLoop(
+                null, engine, 0, 1_000_000L,
+                () -> {
+                    throw new IOException("no reconnect in this test");
+                },
+                0, 0, 1);
+    }
+
     private static void populateRecoverableSlot(Path sfDir) throws Exception {
         try (TestWebSocketServer silent = new TestWebSocketServer(new SilentHandler())) {
             int port = silent.getPort();
@@ -117,6 +181,12 @@ private static void populateRecoverableSlot(Path sfDir) throws Exception {
         }
     }
 
+    private static int readInt(CursorWebSocketSendLoop loop, String name) throws Exception {
+        Field f = CursorWebSocketSendLoop.class.getDeclaredField(name);
+        f.setAccessible(true);
+        return f.getInt(loop);
+    }
+
     private static void rmDir(Path dir) throws IOException {
         if (dir == null || !Files.exists(dir)) {
             return;