Discard a surviving symbol dict on fresh start

A fresh store-and-forward slot inherited a stale .symbol-dict when the
best-effort removeOrphan failed to delete it (e.g. a Windows share lock)
or a crash landed in the close window: open() parses and TRUSTS any
survivor. Because the fresh-start producer is not seeded from the
dictionary (that is gated on wasRecoveredFromDisk, while the send-loop
mirror and the persist resume point key off pd.size()), the producer's
ids then diverged from the dictionary the send loop replays, silently
misattributing symbol values after a reconnect.

The dictionary is load-bearing, unlike the max()-clamped ack watermark
whose removeOrphan+open pattern it had copied. Enforce "fresh start ->
empty dictionary" at the source: openClean() truncates any survivor via
openCleanRW instead of trusting it, and if the clean open itself fails
the sender falls back to full self-sufficient frames, which is also
safe. The recovery path keeps open(); removeOrphan stays for the
fully-drained close.

Both regression tests fail without the fix (dict size 2, not 0):
- PersistedSymbolDictTest.testOpenCleanDiscardsSurvivingDictionary
- EmptyOrphanSlotChurnTest.testFreshStartDiscardsSurvivingStaleDictionary

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/CursorSendEngine.java

@@ -321,10 +321,19 @@ private CursorSendEngine(String sfDir, long segmentSizeBytes, SegmentManager man
                 if (!memoryMode) {
                     AckWatermark.removeOrphan(sfDir);
                     watermarkInProgress = AckWatermark.open(sfDir);
-                    // Same stale-side-file hygiene for the symbol dictionary: a
-                    // fresh slot starts with an empty dictionary.
-                    PersistedSymbolDict.removeOrphan(sfDir);
-                    persistedDictInProgress = PersistedSymbolDict.open(sfDir);
+                    // A fresh slot MUST start with an EMPTY symbol dictionary.
+                    // Unlike the ack watermark above -- a discardable optimization a
+                    // max() clamp protects -- the dictionary is load-bearing: a
+                    // delta frame referencing an id missing from it is unrecoverable,
+                    // and a STALE dictionary inherited here (the segments are gone, so
+                    // the producer is NOT seeded from it) shifts the dense id->symbol
+                    // mapping and silently misattributes symbols on the next
+                    // reconnect. openClean() truncates any survivor to empty rather
+                    // than trusting a best-effort delete that may have failed (e.g. a
+                    // Windows share lock); if the clean open itself fails,
+                    // persistedSymbolDict stays null and the sender falls back to full
+                    // self-sufficient frames, which is also safe.
+                    persistedDictInProgress = PersistedSymbolDict.openClean(sfDir);
                 }
                 MmapSegment initial;
                 String initialPath = null;

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

@@ -147,10 +147,31 @@ public static PersistedSymbolDict open(String slotDir) {
     }
 
     /**
-     * Best-effort removal of a stale dictionary file. Used at fresh-start (a
-     * stale dict with no segments behind it is meaningless) and at fully-drained
-     * close (the slot is empty, nothing references the dictionary any more),
-     * mirroring {@link AckWatermark#removeOrphan}.
+     * Opens the dictionary in {@code slotDir} as a FRESH, EMPTY file, discarding
+     * any surviving content. This is the fresh-start counterpart to {@link #open}:
+     * a slot with no recovered segments must start with an empty dictionary, so a
+     * dictionary left by a prior lifecycle -- a fully-drained slot whose
+     * best-effort delete failed, or a crash in the close window -- must NOT be
+     * inherited. Unlike {@link #open}, which parses and TRUSTS an existing file for
+     * recovery/orphan-drain replay, this truncates it: the fresh-start producer is
+     * not seeded from the dictionary, so trusting a survivor would leave the
+     * producer's ids diverged from the dictionary the send loop replays and
+     * silently misattribute symbols on the next reconnect. Truncating (rather than
+     * relying on an unlink succeeding first) closes the gap even when the delete is
+     * refused -- e.g. a Windows share lock. Returns {@code null} on I/O failure, so
+     * the caller falls back to full self-sufficient frames exactly as {@link #open}
+     * does.
+     */
+    public static PersistedSymbolDict openClean(String slotDir) {
+        return openFresh(slotDir + "/" + FILE_NAME);
+    }
+
+    /**
+     * Best-effort removal of a stale dictionary file. Used at fully-drained close
+     * (the slot is empty, nothing references the dictionary any more), mirroring
+     * {@link AckWatermark#removeOrphan}. The fresh-start path deliberately does NOT
+     * use this -- it opens a clean dictionary via {@link #openClean} instead, so a
+     * failed delete cannot leave a stale dictionary a new session would trust.
      */
     public static void removeOrphan(String slotDir) {
         Files.remove(slotDir + "/" + FILE_NAME);

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

@@ -25,6 +25,7 @@
 package io.questdb.client.test.cutlass.qwp.client.sf.cursor;
 
 import io.questdb.client.cutlass.qwp.client.sf.cursor.CursorSendEngine;
+import io.questdb.client.cutlass.qwp.client.sf.cursor.PersistedSymbolDict;
 import io.questdb.client.std.Files;
 import io.questdb.client.test.tools.TestUtils;
 import org.junit.After;
@@ -35,6 +36,7 @@
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
 
 /**
  * Regression test for M6 — drainer adopting an empty orphan slot would
@@ -87,6 +89,48 @@ public void tearDown() {
         Files.remove(sfDir);
     }
 
+    @Test
+    public void testFreshStartDiscardsSurvivingStaleDictionary() throws Exception {
+        // Regression: a prior fully-drained lifecycle can leave a stale
+        // .symbol-dict behind (a best-effort delete that failed, or a crash in the
+        // close window) with NO segments. A fresh start must DISCARD it -- the
+        // dictionary is load-bearing and the fresh-start producer is not seeded
+        // from it, so trusting a survivor would diverge the producer ids from the
+        // dictionary the send loop replays and misattribute symbols on reconnect.
+        TestUtils.assertMemoryLeak(() -> {
+            // Pre-seed a stale dictionary in the slot, with no segments behind it.
+            PersistedSymbolDict stale = PersistedSymbolDict.open(sfDir);
+            assertNotNull(stale);
+            try {
+                stale.appendSymbol("staleX");
+                stale.appendSymbol("staleY");
+                assertEquals(2, stale.size());
+            } finally {
+                stale.close();
+            }
+
+            // A fresh start (no recovered segments) must open a CLEAN, empty
+            // dictionary -- not inherit the survivor.
+            try (CursorSendEngine engine = new CursorSendEngine(sfDir, 4L * 1024 * 1024)) {
+                assertFalse("fresh start must not report a disk recovery",
+                        engine.wasRecoveredFromDisk());
+                PersistedSymbolDict pd = engine.getPersistedSymbolDict();
+                assertNotNull(pd);
+                assertEquals("fresh start must discard the surviving stale dictionary",
+                        0, pd.size());
+            }
+
+            // The survivor's bytes are physically gone, not just hidden.
+            PersistedSymbolDict reopened = PersistedSymbolDict.open(sfDir);
+            assertNotNull(reopened);
+            try {
+                assertEquals(0, reopened.size());
+            } finally {
+                reopened.close();
+            }
+        });
+    }
+
     @Test
     public void testNeverPublishedCloseLeavesNoSfaFiles() throws Exception {
         TestUtils.assertMemoryLeak(() -> {

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

@@ -299,6 +299,55 @@ public void testLargeSymbolRoundTripsAcrossReopen() throws Exception {
         });
     }
 
+    @Test
+    public void testOpenCleanDiscardsSurvivingDictionary() throws Exception {
+        // A fresh start must NOT inherit a dictionary left by a prior lifecycle:
+        // openClean() truncates any survivor to empty, where open() would recover
+        // (and TRUST) it. Trusting a survivor whose segments are gone -- the
+        // fresh-start producer is not seeded from it -- shifts the dense id->symbol
+        // mapping and misattributes symbols on the next reconnect.
+        assertMemoryLeak(() -> {
+            Path dir = Files.createTempDirectory("qwp-symdict-clean");
+            try {
+                PersistedSymbolDict stale = PersistedSymbolDict.open(dir.toString());
+                Assert.assertNotNull(stale);
+                try {
+                    stale.appendSymbol("staleX");
+                    stale.appendSymbol("staleY");
+                    Assert.assertEquals(2, stale.size());
+                } finally {
+                    stale.close();
+                }
+
+                // Fresh start: openClean yields an EMPTY dictionary regardless of
+                // the survivor, and appends from id 0 again.
+                PersistedSymbolDict fresh = PersistedSymbolDict.openClean(dir.toString());
+                Assert.assertNotNull(fresh);
+                try {
+                    Assert.assertEquals(0, fresh.size());
+                    Assert.assertEquals(0, fresh.readLoadedSymbols().size());
+                    fresh.appendSymbol("freshA");
+                    Assert.assertEquals(1, fresh.size());
+                } finally {
+                    fresh.close();
+                }
+
+                // The survivor's bytes are physically gone, not just hidden: a
+                // subsequent recovery open() sees only the post-clean content.
+                PersistedSymbolDict reopened = PersistedSymbolDict.open(dir.toString());
+                Assert.assertNotNull(reopened);
+                try {
+                    Assert.assertEquals(1, reopened.size());
+                    Assert.assertEquals("freshA", reopened.readLoadedSymbols().getQuick(0));
+                } finally {
+                    reopened.close();
+                }
+            } finally {
+                rmDir(dir);
+            }
+        });
+    }
+
     @Test
     public void testRemoveOrphanDeletesFile() throws Exception {
         assertMemoryLeak(() -> {