Guard torn-dict resume and skip split re-encode

Two hardening fixes to the delta symbol-dictionary path.

Torn-dictionary resume (data integrity): the persisted .symbol-dict is
not fsync'd, so a host crash can lose its newest entries while the
segment frames that introduced those ids survive and replay. A resumed
producer, seeded from the shorter dictionary, would reuse ids the
surviving frames already define and silently misattribute symbols. The
send loop's replay guard only catches a gap (deltaStart > mirror); the
contiguous tear self-heals the mirror and leaves the producer diverged.
CursorSendEngine now records the highest symbol id the recovered frames
reference via a read-only MmapSegment/SegmentRing walk, and
seedGlobalDictionaryFromPersisted fails clean when it reaches the
recovered dictionary size -- the affected data must be resent. Only the
resuming producer refuses; the orphan drainer still self-heals and
drains the slot. testTornDictSubsetFailsCleanOnResume covers it.

Split-flush sizing (performance): flushPendingRowsSplit re-encoded the
whole batch a second time just to size each split frame. It now derives
each frame size arithmetically from the combined encode flushPendingRows
already performed, capturing each table's body bytes and the delta-entry
length. Body encoding is context-free, so the sizes are exact; the
publish loop's assert cross-checks them.

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

Author: glasstiger

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

@@ -57,6 +57,7 @@
 import io.questdb.client.std.Decimal128;
 import io.questdb.client.std.Decimal256;
 import io.questdb.client.std.Decimal64;
+import io.questdb.client.std.IntList;
 import io.questdb.client.std.Misc;
 import io.questdb.client.std.Numbers;
 import io.questdb.client.std.NumericException;
@@ -169,6 +170,11 @@ public class QwpWebSocketSender implements Sender {
     // behind a drainer's endpoint walk.
     private final ReentrantLock connectWalkLock = new ReentrantLock();
     private final QwpHostHealthTracker hostTracker;
+    // Per-table encoded body byte counts captured during flushPendingRows' combined
+    // encode, reused by flushPendingRowsSplit to size each split frame arithmetically
+    // instead of re-encoding the batch a second time. Cleared and repopulated on every
+    // flush; only consumed on the (exceptional) split path. Reused to stay zero-GC.
+    private final IntList splitFrameBodyBytes = new IntList();
     private final CharSequenceObjHashMap<QwpTableBuffer> tableBuffers;
     // null means plain text (no TLS)
     private final ClientTlsConfiguration tlsConfig;
@@ -3474,6 +3480,15 @@ private void flushPendingRows(boolean deferCommit) {
         encoder.setDeferCommit(deferCommit);
         encoder.beginMessage(tableCount, globalSymbolDictionary,
                 symbolDeltaBaseline(), currentBatchMaxSymbolId);
+        // Record each table's encoded body size (position delta across addTable) so
+        // the split path can size its per-table frames arithmetically rather than
+        // re-encoding the whole batch. Body encoding is context-free (a table's bytes
+        // don't depend on its siblings or the delta section), so the size a table
+        // takes here equals the size it takes in its own split frame. Only consumed
+        // when the batch overflows the cap; the capture is a couple of int ops per
+        // table on the common path.
+        splitFrameBodyBytes.clear();
+        int bodyStart = encoder.getBuffer().getPosition();
         for (int i = 0, n = keys.size(); i < n; i++) {
             CharSequence tableName = keys.getQuick(i);
             if (tableName == null) {
@@ -3490,12 +3505,18 @@ private void flushPendingRows(boolean deferCommit) {
             }
 
             encoder.addTable(tableBuffer);
+            int bodyEnd = encoder.getBuffer().getPosition();
+            splitFrameBodyBytes.add(bodyEnd - bodyStart);
+            bodyStart = bodyEnd;
         }
         int messageSize = encoder.finishMessage();
         QwpBufferWriter buffer = encoder.getBuffer();
 
         if (serverMaxBatchSize > 0 && messageSize > serverMaxBatchSize) {
-            flushPendingRowsSplit(keys, deferCommit);
+            // The combined frame's delta-entry bytes are byte-identical to the first
+            // split frame's (same baseline + batch max), so capture the length now
+            // for the arithmetic frame-sizing in flushPendingRowsSplit.
+            flushPendingRowsSplit(keys, deferCommit, encoder.getDeltaEntriesLen());
             return;
         }
 
@@ -3547,7 +3568,7 @@ private void flushPendingRows(boolean deferCommit) {
      *                    carry FLAG_DEFER_COMMIT. When false, only the
      *                    last message omits the flag.
      */
-    private void flushPendingRowsSplit(ObjList<CharSequence> keys, boolean deferCommit) {
+    private void flushPendingRowsSplit(ObjList<CharSequence> keys, boolean deferCommit, int combinedDeltaEntriesLen) {
         if (LOG.isDebugEnabled()) {
             LOG.debug("Splitting flush across multiple messages [serverMaxBatchSize={}, defer={}]", serverMaxBatchSize, deferCommit);
         }
@@ -3561,15 +3582,19 @@ private void flushPendingRowsSplit(ObjList<CharSequence> keys, boolean deferComm
         // resetTableBuffersAfterFlush would discard every source row -- a partial
         // commit the caller was told (by the throw) had failed. Checking all sizes up
         // front makes the split all-or-nothing: either every frame fits and all
-        // publish, or none publish and we throw with nothing stranded. The cost is a
-        // second encode pass over the split batch, which is already the exceptional
-        // large-batch path. encode is read-only on the table buffer, and simBaseline
-        // mirrors the publish loop's baseline advance (advanceSentMaxSymbolId), so
-        // each measured size equals the frame the publish loop will build; this pass
-        // mutates no delta/persist state (the defer-commit flag is a header bit that
-        // does not change frame size).
+        // publish, or none publish and we throw with nothing stranded.
+        //
+        // Each split frame's size is derived ARITHMETICALLY from the combined encode
+        // flushPendingRows already performed -- header + the two delta-section varints
+        // + the delta entries (byte-identical to the combined frame's when this frame
+        // carries them) + the table's own body bytes (captured in splitFrameBodyBytes,
+        // context-free so identical here and in its solo frame) -- rather than
+        // re-encoding every table a second time. simBaseline mirrors the publish loop's
+        // baseline advance (advanceSentMaxSymbolId), so each size equals the frame the
+        // publish loop will build; this pass mutates no delta/persist state.
         int nonEmptyCount = 0;
         int simBaseline = symbolDeltaBaseline();
+        int bodyIdx = 0;
         for (int i = 0, n = keys.size(); i < n; i++) {
             CharSequence tableName = keys.getQuick(i);
             if (tableName == null) {
@@ -3580,9 +3605,14 @@ private void flushPendingRowsSplit(ObjList<CharSequence> keys, boolean deferComm
                 continue;
             }
             nonEmptyCount++;
-            encoder.beginMessage(1, globalSymbolDictionary, simBaseline, currentBatchMaxSymbolId);
-            encoder.addTable(tableBuffer);
-            int messageSize = encoder.finishMessage();
+            int deltaStart = simBaseline + 1;
+            int deltaCount = Math.max(0, currentBatchMaxSymbolId - simBaseline);
+            int messageSize = QwpConstants.HEADER_SIZE
+                    + NativeBufferWriter.varintSize(deltaStart)
+                    + NativeBufferWriter.varintSize(deltaCount)
+                    + (deltaCount > 0 ? combinedDeltaEntriesLen : 0)
+                    + splitFrameBodyBytes.getQuick(bodyIdx);
+            bodyIdx++;
             if (messageSize > serverMaxBatchSize) {
                 resetTableBuffersAfterFlush(keys);
                 throw new LineSenderException("single table batch too large for server batch cap")
@@ -3821,11 +3851,38 @@ private void resetSymbolDictStateForNewConnection() {
      * 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.
+     * <p>
+     * <b>Host-crash tear guard.</b> The persisted dictionary is NOT fsync'd (see
+     * {@code PersistedSymbolDict}), so a host/power crash can lose its
+     * most-recently-written (highest-id) entries while the segment frames that
+     * introduced those ids survive -- and those newest frames, being the least
+     * likely to be acked, replay on recovery. The send loop's catch-up mirror then
+     * rebuilds the missing ids from those frames' own delta bytes, but THIS producer
+     * -- seeded only from the shorter dictionary -- would assign its next new symbol
+     * an id the surviving frames already define, putting two symbols on one id and
+     * silently misattributing values. The send loop's replay guard only catches a
+     * GAP ({@code deltaStart > sentDictCount}); a frame that introduces exactly the
+     * torn-off id ({@code deltaStart == pd.size()}) slips through and self-heals the
+     * mirror, leaving only this producer diverged. Detect it here -- the surviving
+     * frames reference an id at or beyond the recovered dictionary size -- and fail
+     * clean: the affected data must be resent, matching the design's torn-dict
+     * "resend required" contract. The recovered data is not lost; the background
+     * drainer still drains this slot (its mirror self-heals from the frames). Only a
+     * host crash reaches this -- a process crash keeps the page cache, so the
+     * write-ahead ordering keeps the dictionary a superset of the frames.
      */
     private void seedGlobalDictionaryFromPersisted(PersistedSymbolDict pd) {
         if (pd == null || pd.size() == 0) {
             return;
         }
+        if (cursorEngine != null && cursorEngine.recoveredMaxSymbolId() >= pd.size()) {
+            throw new LineSenderException(
+                    "recovered store-and-forward symbol dictionary is a subset of the surviving frames "
+                            + "(likely a host crash tore its unsynced tail): frames reference symbol id "
+                            + cursorEngine.recoveredMaxSymbolId() + " but the recovered dictionary holds only "
+                            + pd.size() + " id(s); resuming would reuse ids the frames already define -- "
+                            + "resend the affected data");
+        }
         ObjList<String> symbols = pd.readLoadedSymbols();
         for (int i = 0, n = symbols.size(); i < n; i++) {
             globalSymbolDictionary.addRecoveredSymbol(symbols.getQuick(i));

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

@@ -96,6 +96,15 @@ public final class CursorSendEngine implements QuietCloseable {
     // covers them. Read by the sender's close-time drain to avoid waiting on
     // acks that cannot arrive.
     private long recoveredCommitBoundaryFsn = -1L;
+    // Highest symbol id any recovered delta frame references, or -1 for
+    // fresh/memory rings (and recovered rings with no symbol-bearing frame). A
+    // resuming producer seeds its dictionary baseline from the persisted
+    // .symbol-dict; if that dictionary was torn below this id by a host crash
+    // (the side-file is not fsync'd), the producer would re-use ids the surviving
+    // frames already define. seedGlobalDictionaryFromPersisted compares this
+    // against the recovered dictionary size to fail clean instead. Computed once
+    // in the constructor's recovery branch; -1 elsewhere.
+    private long recoveredMaxSymbolId = -1L;
     // FSN of the last frame of a recovered orphaned deferred tail, or -1 when
     // the recovered ring has no such tail. When >= 0, frames
     // [recoveredCommitBoundaryFsn + 1 .. recoveredOrphanTipFsn] all carry
@@ -312,6 +321,18 @@ 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.
+                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;
             } else {
                 // Fresh start with no recovered segments. Any stale
                 // watermark from a prior fully-drained session refers
@@ -714,6 +735,16 @@ public long recoveredCommitBoundaryFsn() {
         return recoveredCommitBoundaryFsn;
     }
 
+    /**
+     * Highest symbol id any recovered delta frame references, or {@code -1} for
+     * fresh/memory rings (and recovered rings with no symbol-bearing frame). A
+     * resuming producer compares this against its recovered dictionary size to
+     * detect a host-crash tear of the persisted {@code .symbol-dict}.
+     */
+    public long recoveredMaxSymbolId() {
+        return recoveredMaxSymbolId;
+    }
+
     /**
      * FSN of the last frame of a recovered orphaned deferred tail, or
      * {@code -1} when none. See {@link #recoveredCommitBoundaryFsn()}: the

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

@@ -522,6 +522,72 @@ public long findLastFrameFsnWithoutPayloadFlag(int flagsOffset, int flagMask, in
         return best;
     }
 
+    /**
+     * Highest {@code deltaStart + deltaCount} (one past the highest symbol id) that
+     * any symbol-dict delta frame in this segment references, or {@code 0} when no
+     * such frame carries a symbol. Read-only walk over the recovered frames, used
+     * once at recovery to detect a persisted {@code .symbol-dict} torn (host crash,
+     * 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]...}.
+     *
+     * @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)
+     */
+    public long maxSymbolDeltaEnd(int headerMagic, int flagsOffset, int flagDeltaMask, int qwpHeaderSize) {
+        long maxEnd = 0L;
+        long off = HEADER_SIZE;
+        long frames = frameCount;
+        for (long i = 0; i < frames; i++) {
+            int payloadLen = Unsafe.getUnsafe().getInt(mmapAddress + off + 4);
+            long payload = mmapAddress + off + FRAME_HEADER_SIZE;
+            if (payloadLen >= qwpHeaderSize
+                    && payloadLen > flagsOffset
+                    && Unsafe.getUnsafe().getInt(payload) == headerMagic
+                    && (Unsafe.getUnsafe().getByte(payload + flagsOffset) & flagDeltaMask) != 0) {
+                long p = payload + qwpHeaderSize;
+                long limit = payload + payloadLen;
+                long deltaStart = 0L;
+                int shift = 0;
+                while (p < limit) {
+                    byte b = Unsafe.getUnsafe().getByte(p++);
+                    deltaStart |= (long) (b & 0x7F) << shift;
+                    if ((b & 0x80) == 0) {
+                        break;
+                    }
+                    shift += 7;
+                    if (shift > 35) {
+                        break; // corrupt run; recovered frames are CRC-valid, so defensive only
+                    }
+                }
+                long deltaCount = 0L;
+                shift = 0;
+                while (p < limit) {
+                    byte b = Unsafe.getUnsafe().getByte(p++);
+                    deltaCount |= (long) (b & 0x7F) << shift;
+                    if ((b & 0x80) == 0) {
+                        break;
+                    }
+                    shift += 7;
+                    if (shift > 35) {
+                        break;
+                    }
+                }
+                long end = deltaStart + deltaCount;
+                if (end > maxEnd) {
+                    maxEnd = end;
+                }
+            }
+            off += FRAME_HEADER_SIZE + payloadLen;
+        }
+        return maxEnd;
+    }
+
     /**
      * Number of frames written since {@link #create} (or recovered by
      * {@link #openExisting}). Used by {@code SegmentRing} to compute

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

@@ -546,6 +546,24 @@ public synchronized long findLastFsnWithoutPayloadFlag(int flagsOffset, int flag
         return Math.max(best, fsn);
     }
 
+    /**
+     * 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.
+     */
+    public synchronized long maxSymbolDeltaEnd(int headerMagic, int flagsOffset, int flagDeltaMask, int qwpHeaderSize) {
+        long maxEnd = 0L;
+        for (int i = 0, n = sealedSegments.size(); i < n; i++) {
+            long end = sealedSegments.get(i).maxSymbolDeltaEnd(headerMagic, flagsOffset, flagDeltaMask, qwpHeaderSize);
+            if (end > maxEnd) {
+                maxEnd = end;
+            }
+        }
+        long end = active.maxSymbolDeltaEnd(headerMagic, flagsOffset, flagDeltaMask, qwpHeaderSize);
+        return Math.max(maxEnd, end);
+    }
+
     public MmapSegment getActive() {
         return active;
     }