Fix recovery id desync on UTF-8-colliding symbols

Two DISTINCT source symbols that collapse to the same UTF-8 bytes -- lone
UTF-16 surrogates, which the encoder maps to '?' -- get distinct producer
ids and persist as separate .symbol-dict entries. On recovery,
seedGlobalDictionaryFromPersisted replayed them through getOrAddSymbol,
which de-dups the decoded "?" strings, leaving the producer dictionary
(and sentMaxSymbolId) one short of pd.size(). That desynced the delta
baseline from the send-loop catch-up mirror (which uses pd.size()) and,
after a reconnect, silently misattributed later well-formed symbols.

Add GlobalSymbolDictionary.addRecoveredSymbol, which appends at the next
id WITHOUT de-duplicating, and seed recovery through it so the producer
id space always matches the persisted entry count. The reverse lookup
keeps the highest id for a colliding string, which is harmless -- both
ids encode to identical bytes.

A GlobalSymbolDictionary unit test pins the no-dedup contract, and a new
recovery test seeds a slot with two lone-surrogate symbols and asserts
the producer dictionary and sentMaxSymbolId match the persisted entry
count. It fails before this change (dictionary size 1 vs the file's 2).

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

Author: glasstiger

core/src/main/java/io/questdb/client/cutlass/qwp/client/GlobalSymbolDictionary.java

@@ -50,6 +50,37 @@ public GlobalSymbolDictionary(int initialCapacity) {
         this.idToSymbol = new ObjList<>(initialCapacity);
     }
 
+    /**
+     * Appends {@code symbol} at the next sequential id, matching a recovered /
+     * persisted dictionary's dense id order, WITHOUT de-duplicating.
+     * <p>
+     * Recovery ({@code QwpWebSocketSender.seedGlobalDictionaryFromPersisted})
+     * replays the persisted entries in id order to rebuild this dictionary. It must
+     * NOT collapse two source strings that decode to the same characters, because
+     * the persisted {@code .symbol-dict}, the on-wire delta and the I/O-thread
+     * catch-up mirror all key on the entry POSITION (id), not on the string. The
+     * only strings that collide this way are malformed lone UTF-16 surrogates,
+     * which the UTF-8 encoder maps to {@code '?'}: {@link #getOrAddSymbol} would
+     * de-dup them and leave this dictionary SHORTER than the persisted entry count,
+     * desyncing the producer's delta baseline from the catch-up mirror (which uses
+     * {@code pd.size()}) and silently misattributing later symbols. Appending
+     * unconditionally keeps {@link #size()} equal to that count. The reverse lookup
+     * keeps the highest id for a colliding string, which is harmless: both ids
+     * encode to the same bytes, so resolving either is equivalent.
+     *
+     * @param symbol the recovered symbol string (must not be null)
+     * @return the id assigned (the previous {@link #size()})
+     */
+    public int addRecoveredSymbol(String symbol) {
+        if (symbol == null) {
+            throw new IllegalArgumentException("symbol cannot be null");
+        }
+        int newId = idToSymbol.size();
+        symbolToId.put(symbol, newId);
+        idToSymbol.add(symbol);
+        return newId;
+    }
+
     /**
      * Clears all symbols from the dictionary.
      * <p>

core/src/main/java/io/questdb/client/cutlass/qwp/client/QwpWebSocketSender.java

@@ -3750,14 +3750,24 @@ private void resetSymbolDictStateForNewConnection() {
      * the slot's persisted dictionary (ids assigned in the same ascending order,
      * so they match the recovered frames) and resumes the delta baseline at the
      * recovered tip, so newly ingested symbols continue above the recovered ids.
+     * <p>
+     * Uses {@link GlobalSymbolDictionary#addRecoveredSymbol} (append, NOT de-dup):
+     * the persisted dictionary, the on-wire delta and the send-loop catch-up mirror
+     * all key on the entry POSITION (id), so the producer id space must match the
+     * persisted entry count exactly. {@code getOrAddSymbol} would collapse two
+     * source strings that decode to the same characters -- only malformed lone
+     * UTF-16 surrogates, which UTF-8-encode to {@code '?'} -- leaving this
+     * dictionary shorter than {@code pd.size()} and desyncing
+     * {@code sentMaxSymbolId} from the mirror's {@code sentDictCount = pd.size()},
+     * which silently misattributes later symbols after a reconnect.
      */
     private void seedGlobalDictionaryFromPersisted(PersistedSymbolDict pd) {
         if (pd == null || pd.size() == 0) {
             return;
         }
         ObjList<String> symbols = pd.readLoadedSymbols();
         for (int i = 0, n = symbols.size(); i < n; i++) {
-            globalSymbolDictionary.getOrAddSymbol(symbols.getQuick(i));
+            globalSymbolDictionary.addRecoveredSymbol(symbols.getQuick(i));
         }
         sentMaxSymbolId = globalSymbolDictionary.size() - 1;
     }

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

@@ -36,6 +36,7 @@
 import org.junit.Test;
 
 import java.io.IOException;
+import java.lang.reflect.Field;
 import java.nio.ByteBuffer;
 import java.nio.ByteOrder;
 import java.nio.charset.StandardCharsets;
@@ -713,6 +714,76 @@ public void testRecoveryAfterFailedPublishReplaysGapFree() throws Exception {
         });
     }
 
+    @Test
+    public void testRecoverySeedKeepsUtf8CollidingSymbolsInLockstep() throws Exception {
+        // M2 regression: two DISTINCT source symbols that collapse to the SAME UTF-8
+        // bytes -- lone UTF-16 surrogates, which the encoder maps to '?' -- get
+        // distinct producer ids and persist as separate entries. On recovery the
+        // producer must rebuild its id space to match the persisted entry count
+        // exactly. Pre-fix, seedGlobalDictionaryFromPersisted used getOrAddSymbol,
+        // which de-duped the two decoded "?" strings, leaving the producer
+        // dictionary (and sentMaxSymbolId) one short of pd.size() -- desyncing from
+        // the send-loop catch-up mirror (which uses pd.size()) and silently
+        // misattributing later symbols after a reconnect. addRecoveredSymbol appends
+        // without de-duping, keeping producer and mirror in lockstep.
+        assertMemoryLeak(() -> {
+            // Phase 1: ingest two lone-surrogate symbols in file mode, close-fast.
+            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 s1 = Sender.fromConfig(cfg)) {
+                    s1.table("m").symbol("s", "\uD800").longColumn("v", 1L).atNow(); // lone surrogate -> '?'
+                    s1.flush();
+                    s1.table("m").symbol("s", "\uD801").longColumn("v", 2L).atNow(); // a DIFFERENT one -> '?'
+                    s1.flush();
+                }
+            }
+
+            // The persisted dictionary holds TWO entries (both encode to '?').
+            int persistedSize;
+            PersistedSymbolDict pd = PersistedSymbolDict.open(Paths.get(sfDir, "default").toString());
+            Assert.assertNotNull(pd);
+            try {
+                persistedSize = pd.size();
+            } finally {
+                pd.close();
+            }
+            Assert.assertEquals("two colliding symbols must persist as two entries", 2, persistedSize);
+
+            // Phase 2: recover. The seeded producer id space must match pd.size().
+            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)) {
+                    Assert.assertEquals("recovered producer dictionary must match the persisted "
+                                    + "entry count (not de-duped), else the delta baseline desyncs from the mirror",
+                            persistedSize, globalDictSize(s2));
+                    Assert.assertEquals("delta baseline must resume at the persisted tip",
+                            persistedSize - 1, intField(s2, "sentMaxSymbolId"));
+                }
+            }
+        });
+    }
+
+    private static int globalDictSize(Sender sender) throws Exception {
+        Field f = sender.getClass().getDeclaredField("globalSymbolDictionary");
+        f.setAccessible(true);
+        Object dict = f.get(sender);
+        return (int) dict.getClass().getMethod("size").invoke(dict);
+    }
+
+    private static int intField(Sender sender, String name) throws Exception {
+        Field f = sender.getClass().getDeclaredField(name);
+        f.setAccessible(true);
+        return f.getInt(sender);
+    }
+
     private static void writeAckWatermark(java.nio.file.Path path, long fsn) throws IOException {
         byte[] buf = new byte[16];
         ByteBuffer bb = ByteBuffer.wrap(buf).order(ByteOrder.LITTLE_ENDIAN);

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

@@ -31,6 +31,36 @@
 
 public class GlobalSymbolDictionaryTest {
 
+    @Test
+    public void testAddRecoveredSymbol_appendsWithoutDeduplicating() {
+        // Recovery replays persisted entries in id order. Distinct source strings
+        // that decode to the same characters -- lone UTF-16 surrogates both
+        // UTF-8-encode to '?', so they read back as the string "?" -- must keep
+        // DISTINCT ids, so the producer id space matches the persisted entry count.
+        // getOrAddSymbol de-dups them; addRecoveredSymbol must not.
+        GlobalSymbolDictionary dedup = new GlobalSymbolDictionary();
+        dedup.getOrAddSymbol("?");
+        dedup.getOrAddSymbol("?");
+        assertEquals("getOrAddSymbol de-dups colliding strings", 1, dedup.size());
+
+        GlobalSymbolDictionary recovered = new GlobalSymbolDictionary();
+        assertEquals(0, recovered.addRecoveredSymbol("?"));
+        assertEquals(1, recovered.addRecoveredSymbol("?"));
+        assertEquals(2, recovered.addRecoveredSymbol("nvda"));
+        assertEquals("addRecoveredSymbol keeps colliding entries distinct", 3, recovered.size());
+
+        // Dense id -> symbol mapping is preserved position-for-position.
+        assertEquals("?", recovered.getSymbol(0));
+        assertEquals("?", recovered.getSymbol(1));
+        assertEquals("nvda", recovered.getSymbol(2));
+
+        // A later ingest of a colliding string reuses the highest recovered id
+        // (harmless -- both encode to identical bytes), and a genuinely new symbol
+        // continues past the recovered tip.
+        assertEquals(1, recovered.getOrAddSymbol("?"));
+        assertEquals(3, recovered.getOrAddSymbol("brand-new"));
+    }
+
     @Test
     public void testAddSymbol_assignsSequentialIds() {
         GlobalSymbolDictionary dict = new GlobalSymbolDictionary();