Bound recovered symbol id to committed frames

CursorSendEngine.recoveredMaxSymbolId walked every recovered frame,
including the aborted orphan-deferred tail -- the frames trySendOne
retires without ever transmitting. Those frames typically introduce the
highest symbol ids, so a host crash that tore the persisted .symbol-dict
down to the committed ids while an orphan-tail frame referenced a higher
one inflated recoveredMaxSymbolId and made seedGlobalDictionaryFromPersisted
reject an otherwise fully recoverable slot. The producer never reuses an
orphan id on the wire -- the tail retires before its new frames send -- so
the rejection was a false positive.

Bound the maxSymbolDeltaEnd walk to recoveredCommitBoundaryFsn so only
committed frames count. MmapSegment and SegmentRing take a maxFsnInclusive
argument; a frame whose fsn (baseSeq + i) exceeds it is skipped, and the
walk still advances past it. A slot with no orphan tail is unchanged: the
boundary equals publishedFsn, so every frame still counts.

Add testRecoveredMaxSymbolIdExcludesOrphanTailFrames: a committed delta
frame registering ids 0,1 plus a deferred (orphan) frame registering id 2
must recover recoveredMaxSymbolId == 1, not 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/sf/cursor/CursorSendEngine.java

@@ -321,18 +321,22 @@ private CursorSendEngine(String sfDir, long segmentSizeBytes, SegmentManager man
                             publishedFsn - Math.max(recoveredCommitBoundaryFsn, -1L),
                             recoveredCommitBoundaryFsn, publishedFsn);
                 }
-                // Highest symbol id the surviving frames reference. A resuming
-                // producer compares this against its recovered dictionary size
-                // (seedGlobalDictionaryFromPersisted) to detect a host-crash tear:
-                // if a frame references an id the (unsynced, torn) .symbol-dict no
-                // longer holds, resuming would re-use it. maxSymbolDeltaEnd returns
-                // 0 when no frame carries a symbol, yielding -1 here. Computed
-                // before the I/O loop or producer append; single-threaded here.
+                // Highest symbol id the surviving COMMITTED frames reference. A
+                // resuming producer compares this against its recovered dictionary
+                // size (seedGlobalDictionaryFromPersisted) to detect a host-crash
+                // tear: if a committed frame references an id the (unsynced, torn)
+                // .symbol-dict no longer holds, resuming would re-use it. The walk is
+                // bounded to recoveredCommitBoundaryFsn so the aborted orphan-deferred
+                // tail -- retired without ever being transmitted -- does not inflate
+                // this and over-reject an otherwise-recoverable slot. maxSymbolDeltaEnd
+                // returns 0 when no such frame carries a symbol, yielding -1 here.
+                // Computed before the I/O loop or producer append; single-threaded.
                 this.recoveredMaxSymbolId = recovered.maxSymbolDeltaEnd(
                         io.questdb.client.cutlass.qwp.protocol.QwpConstants.MAGIC_MESSAGE,
                         io.questdb.client.cutlass.qwp.protocol.QwpConstants.HEADER_OFFSET_FLAGS,
                         io.questdb.client.cutlass.qwp.protocol.QwpConstants.FLAG_DELTA_SYMBOL_DICT,
-                        io.questdb.client.cutlass.qwp.protocol.QwpConstants.HEADER_SIZE) - 1L;
+                        io.questdb.client.cutlass.qwp.protocol.QwpConstants.HEADER_SIZE,
+                        recoveredCommitBoundaryFsn) - 1L;
             } else {
                 // Fresh start with no recovered segments. Any stale
                 // watermark from a prior fully-drained session refers

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

@@ -530,23 +530,34 @@ public long findLastFrameFsnWithoutPayloadFlag(int flagsOffset, int flagMask, in
      * out-of-order page loss) below the ids the surviving frames still reference: if
      * the max here reaches at or beyond the recovered dictionary size, a resuming
      * producer -- seeded from the shorter dictionary -- would re-use ids the frames
-     * already define. The frame layout mirrors {@link #findLastFrameFsnWithoutPayloadFlag}:
-     * the QWP message header ({@code qwpHeaderSize} bytes) is followed by the delta
-     * section {@code [deltaStart varint][deltaCount varint]...}.
+     * already define. Only frames at or below {@code maxFsnInclusive} count: frames
+     * above it are the aborted orphan-deferred tail, which {@code trySendOne} retires
+     * without ever transmitting, so their ids must not inflate the result (a resuming
+     * producer never reuses them on the wire). The frame layout mirrors
+     * {@link #findLastFrameFsnWithoutPayloadFlag}: the QWP message header
+     * ({@code qwpHeaderSize} bytes) is followed by the delta section
+     * {@code [deltaStart varint][deltaCount varint]...}.
      *
-     * @param headerMagic   the QWP message magic identifying a well-formed frame
-     * @param flagsOffset   byte offset of the flags field within the QWP header
-     * @param flagDeltaMask the FLAG_DELTA_SYMBOL_DICT bit
-     * @param qwpHeaderSize the QWP message header size (delta section starts past it)
+     * @param headerMagic     the QWP message magic identifying a well-formed frame
+     * @param flagsOffset     byte offset of the flags field within the QWP header
+     * @param flagDeltaMask   the FLAG_DELTA_SYMBOL_DICT bit
+     * @param qwpHeaderSize   the QWP message header size (delta section starts past it)
+     * @param maxFsnInclusive highest frame FSN to consider; frames above it are the
+     *                        retired orphan-deferred tail -- pass
+     *                        {@code recoveredCommitBoundaryFsn}
      */
-    public long maxSymbolDeltaEnd(int headerMagic, int flagsOffset, int flagDeltaMask, int qwpHeaderSize) {
+    public long maxSymbolDeltaEnd(int headerMagic, int flagsOffset, int flagDeltaMask, int qwpHeaderSize, long maxFsnInclusive) {
         long maxEnd = 0L;
         long off = HEADER_SIZE;
         long frames = frameCount;
         for (long i = 0; i < frames; i++) {
+            long fsn = baseSeq + i;
             int payloadLen = Unsafe.getUnsafe().getInt(mmapAddress + off + 4);
             long payload = mmapAddress + off + FRAME_HEADER_SIZE;
-            if (payloadLen >= qwpHeaderSize
+            // Skip the orphan-deferred tail (fsn > maxFsnInclusive): retired, never
+            // sent, so its ids must not count. off still advances below.
+            if (fsn <= maxFsnInclusive
+                    && payloadLen >= qwpHeaderSize
                     && payloadLen > flagsOffset
                     && Unsafe.getUnsafe().getInt(payload) == headerMagic
                     && (Unsafe.getUnsafe().getByte(payload + flagsOffset) & flagDeltaMask) != 0) {

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

@@ -547,20 +547,22 @@ public synchronized long findLastFsnWithoutPayloadFlag(int flagsOffset, int flag
     }
 
     /**
-     * Highest {@code deltaStart + deltaCount} any symbol-dict delta frame in the
-     * ring references (0 when none). See {@link MmapSegment#maxSymbolDeltaEnd};
-     * used once at recovery to detect a persisted dictionary torn below the ids
-     * the surviving frames reference.
+     * Highest {@code deltaStart + deltaCount} any symbol-dict delta frame at or below
+     * {@code maxFsnInclusive} references (0 when none). See
+     * {@link MmapSegment#maxSymbolDeltaEnd}; used once at recovery to detect a
+     * persisted dictionary torn below the ids the surviving committed frames
+     * reference. Frames above {@code maxFsnInclusive} are the retired orphan-deferred
+     * tail and are excluded.
      */
-    public synchronized long maxSymbolDeltaEnd(int headerMagic, int flagsOffset, int flagDeltaMask, int qwpHeaderSize) {
+    public synchronized long maxSymbolDeltaEnd(int headerMagic, int flagsOffset, int flagDeltaMask, int qwpHeaderSize, long maxFsnInclusive) {
         long maxEnd = 0L;
         for (int i = 0, n = sealedSegments.size(); i < n; i++) {
-            long end = sealedSegments.get(i).maxSymbolDeltaEnd(headerMagic, flagsOffset, flagDeltaMask, qwpHeaderSize);
+            long end = sealedSegments.get(i).maxSymbolDeltaEnd(headerMagic, flagsOffset, flagDeltaMask, qwpHeaderSize, maxFsnInclusive);
             if (end > maxEnd) {
                 maxEnd = end;
             }
         }
-        long end = active.maxSymbolDeltaEnd(headerMagic, flagsOffset, flagDeltaMask, qwpHeaderSize);
+        long end = active.maxSymbolDeltaEnd(headerMagic, flagsOffset, flagDeltaMask, qwpHeaderSize, maxFsnInclusive);
         return Math.max(maxEnd, end);
     }
 

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

@@ -75,7 +75,9 @@
 public class CursorWebSocketSendLoopOrphanTailTest {
 
     private static final int FLAG_DEFER_COMMIT = 0x01;
+    private static final int FLAG_DELTA_SYMBOL_DICT = 0x08;
     private static final int HEADER_OFFSET_FLAGS = 5;
+    private static final int HEADER_SIZE = 12;
     private static final int MAGIC_MESSAGE = 0x31505751; // "QWP1" little-endian
 
     private String tmpDir;
@@ -221,6 +223,36 @@ public void testSlowPathReplaysBelowTailThenRetiresAndRecyclesOnce() throws Exce
         });
     }
 
+    @Test
+    public void testRecoveredMaxSymbolIdExcludesOrphanTailFrames() throws Exception {
+        // recoveredMaxSymbolId must reflect only COMMITTED (transmitted) frames, not
+        // the aborted orphan-tail frames trySendOne retires without ever sending. A
+        // host crash that tears the persisted dictionary down to the committed ids
+        // while an orphan-tail frame introduced a HIGHER id must NOT over-reject the
+        // resume: the producer never reuses an orphan id on the wire (the tail retires
+        // first), so counting it would inflate recoveredMaxSymbolId and make
+        // seedGlobalDictionaryFromPersisted fail a fully-recoverable slot. The
+        // maxSymbolDeltaEnd walk is therefore bounded to recoveredCommitBoundaryFsn.
+        TestUtils.assertMemoryLeak(() -> {
+            try (CursorSendEngine engine = newEngine()) {
+                // fsn 0: commit-bearing delta frame registering ids 0,1.
+                appendDeltaFrame(engine, false, 0, 2);
+                // fsn 1: DEFERRED delta frame registering id 2 -- the orphan tail.
+                appendDeltaFrame(engine, true, 2, 1);
+            }
+            try (CursorSendEngine engine = newEngine()) {
+                assertTrue(engine.wasRecoveredFromDisk());
+                assertEquals("last commit-bearing frame", 0L, engine.recoveredCommitBoundaryFsn());
+                assertEquals("orphan tail tip", 1L, engine.recoveredOrphanTipFsn());
+                // Only the committed frame's ids (0,1) count -> highest id 1. The
+                // orphan-tail frame's id 2 is excluded, so a resume whose recovered
+                // dictionary holds ids 0,1 (size 2) is NOT over-rejected.
+                assertEquals("orphan-tail id 2 must be excluded from recoveredMaxSymbolId",
+                        1L, engine.recoveredMaxSymbolId());
+            }
+        });
+    }
+
     // ---------------------------------------------------------------------
     // harness
     // ---------------------------------------------------------------------
@@ -291,6 +323,27 @@ private static void appendFrame(CursorSendEngine engine, boolean defer) {
         }
     }
 
+    // Appends a QWP frame carrying a symbol-dict delta section ([deltaStart varint]
+    // [deltaCount varint]) so the recovery walk's maxSymbolDeltaEnd counts it.
+    // deltaStart/deltaCount stay < 128 so each encodes in a single LEB128 byte.
+    private static void appendDeltaFrame(CursorSendEngine engine, boolean defer, int deltaStart, int deltaCount) {
+        int size = HEADER_SIZE + 2;
+        long buf = Unsafe.malloc(size, MemoryTag.NATIVE_DEFAULT);
+        try {
+            for (int i = 0; i < size; i++) {
+                Unsafe.getUnsafe().putByte(buf + i, (byte) 0);
+            }
+            Unsafe.getUnsafe().putInt(buf, MAGIC_MESSAGE);
+            Unsafe.getUnsafe().putByte(buf + HEADER_OFFSET_FLAGS,
+                    (byte) (FLAG_DELTA_SYMBOL_DICT | (defer ? FLAG_DEFER_COMMIT : 0)));
+            Unsafe.getUnsafe().putByte(buf + HEADER_SIZE, (byte) deltaStart);
+            Unsafe.getUnsafe().putByte(buf + HEADER_SIZE + 1, (byte) deltaCount);
+            engine.appendBlocking(buf, size);
+        } finally {
+            Unsafe.free(buf, size, MemoryTag.NATIVE_DEFAULT);
+        }
+    }
+
     private static void awaitAckedFsn(CursorSendEngine engine, long target) throws InterruptedException {
         long deadline = System.nanoTime() + 10_000_000_000L;
         while (engine.ackedFsn() < target) {