Adopt recovered dictionary into the send mirror

On recovery the send loop copied the persisted dictionary's loaded-entries
buffer into a fresh mirror allocation and left PersistedSymbolDict holding a
second copy for the engine's lifetime -- roughly twice the dictionary size in
native memory on a high-cardinality recovered slot, retained long after the
one-time seed. The loop now adopts that buffer as its mirror backing via
takeLoadedEntries(), which transfers ownership so the dictionary no longer
retains or frees it. The producer's readLoadedSymbols() is the only other
consumer and runs first (setCursorEngine seeds the producer before the loop
is built; the drainer has no producer consumer), guarded by an assert.

Add a recover-then-continue-ingest test. A file-mode sender writes symbols
and crashes; a fresh sender recovers the slot and ingests a NEW symbol. It
asserts the producer continues the dictionary from the recovered size instead
of colliding at id 0, exercising seedGlobalDictionaryFromPersisted, which no
prior test drove past recovery.

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,8 +543,16 @@ public CursorWebSocketSendLoop(WebSocketClient client, CursorSendEngine engine,
             if (pd != null && pd.size() > 0) {
                 int len = pd.loadedEntriesLen();
                 if (len > 0) {
-                    ensureSentDictCapacity(len);
-                    Unsafe.getUnsafe().copyMemory(pd.loadedEntriesAddr(), sentDictBytesAddr, len);
+                    // 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();
+                    sentDictBytesCapacity = len;
                     sentDictBytesLen = len;
                     // Set the count only alongside the bytes so sentDictCount can
                     // never claim symbols the mirror does not hold. A recovered slot

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

@@ -94,14 +94,19 @@ public final class PersistedSymbolDict implements QuietCloseable {
     private static final int MAX_ENTRY_LEN = 1 << 20;
     private static final Logger LOG = LoggerFactory.getLogger(PersistedSymbolDict.class);
     private final int fd;
-    // 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 send loop's catch-up mirror and the producer's id map.
-    private final long loadedEntriesAddr;
-    private final int loadedEntriesLen;
     private long appendOffset;
     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.
+    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;
@@ -256,6 +261,7 @@ 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;
@@ -286,6 +292,25 @@ 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/DeltaDictRecoveryTest.java

@@ -227,6 +227,62 @@ public void testTornDictionaryFailsCleanlyInsteadOfCorrupting() throws Exception
         });
     }
 
+    @Test
+    public void testRecoveredSenderContinuesIngestingNewSymbols() throws Exception {
+        // M2 regression: seedGlobalDictionaryFromPersisted resumes the producer's
+        // dictionary and delta baseline from the persisted .symbol-dict, so a
+        // recovered sender that continues ingesting assigns the NEXT id, not a
+        // colliding low one. Without it the new symbol reuses a recovered id and the
+        // fresh server sees a redefinition -> silent misattribution. No prior test
+        // ingests on the recovered sender. Replay is from FSN 0 (no acks), so the
+        // recovered frames legitimately overlap the seeded dictionary -- this also
+        // pins that the redefinition guard does not false-positive on normal
+        // recovery.
+        assertMemoryLeak(() -> {
+            // Phase 1: ingest DISTINCT_SYMBOLS symbols, silent server, close-fast ->
+            // unacked frames + a full persisted dictionary.
+            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
+                        + ";sf_max_bytes=4096;close_flush_timeout_millis=0;";
+                try (Sender s1 = Sender.fromConfig(cfg)) {
+                    for (int i = 0; i < DISTINCT_SYMBOLS; i++) {
+                        s1.table("m").symbol("s", "sym-" + i).longColumn("v", i).atNow();
+                        s1.flush();
+                    }
+                }
+            }
+
+            // Phase 2: recover against a fresh server, then ingest a genuinely NEW
+            // symbol. The producer must continue at id DISTINCT_SYMBOLS.
+            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 + ";";
+                try (Sender s2 = Sender.fromConfig(cfg)) {
+                    s2.table("m").symbol("s", "brand-new").longColumn("v", 99L).atNow();
+                    s2.flush();
+                    long deadline = System.currentTimeMillis() + 10_000;
+                    while (System.currentTimeMillis() < deadline
+                            && handler.maxDictSize() < DISTINCT_SYMBOLS + 1) {
+                        Thread.sleep(20);
+                    }
+                }
+                List<String> dict = handler.dictSnapshot();
+                Assert.assertEquals("recovered sender must continue the dictionary, not collide: " + dict,
+                        DISTINCT_SYMBOLS + 1, dict.size());
+                for (int i = 0; i < DISTINCT_SYMBOLS; i++) {
+                    Assert.assertEquals("sym-" + i, dict.get(i));
+                }
+                Assert.assertEquals("brand-new", dict.get(DISTINCT_SYMBOLS));
+            }
+        });
+    }
+
     @Test
     public void testUnopenablePersistedDictStillGuardsAgainstReplayingDeltaFrames() throws Exception {
         // C1 regression: when a recovered disk slot's persisted dictionary cannot be