Speed up recovery lookups and the per-symbol varint decode

C14. findSegmentContaining walked the sealed list linearly, on the
reconnect path, under the ring monitor -- with the list bounded only by
the documented ~16K-segment ceiling, and its sibling nextSealedAfter
doing the identical predicate in log N right beside it. The list is
sorted by baseSeq with disjoint ranges, so the same binary search
applies. A comparison probe makes it observable, and the test asserts
the counter is non-zero BEFORE asserting the bound: the counter exists
only in the binary search, so a bound-only assertion would pass for free
against a reverted linear scan -- the trap the sibling perf assertions
in this file already fall into.

C15. seedGlobalDictionaryFromPersisted poured the recovered symbols into
a default-capacity (64) dictionary, rehashing the map ~log2(n/64) times,
each pass O(current size). recoveredMaxSymbolId + 1 bounds the seed, so
the dictionary is now pre-sized before the pour. The field is no longer
final; nothing retains a reference -- the encoder and the persisted
dictionary both take it per call -- and the swap happens during
construction, before the producer or the I/O thread exist.

C16. ensureAppendMap rounded the window START down to a 4 MiB boundary
while sizing it to the record's END, so a chunk straddling that boundary
produced a window spanning both: 8 MiB mapped and re-allocated, whose
lower half covers bytes already written and never touched again. Steady
state then advanced 4 MiB per remap while always mapping 8. mmap only
requires a page-aligned offset, so the window is now page-aligned.

C17. readVarintAt returned its end position through an instance field --
a store per call, and this runs once per SYMBOL: per frame in
accumulateSentDict and once per mirror entry in sendDictCatchUp. It now
packs (value << 3) | bytes into the return, the shape
RecoveredFrameAnalysis.readVarint already uses.

That also closes a guard that could not guard. The field version returned
0 with varintEnd == p when p >= limit, so a caller computing
p = varintEnd + len got p == limit and its `p > limit` bail-out could not
fire at an exact boundary; a frame declaring more entries than it carries
would walk zero-length pseudo-entries and advance sentDictCount past
bytes the mirror does not hold. A -1 return makes truncation detectable
and every bail-out now fires.

C13 is RETRACTED, not fixed. The review claimed trySendOne's in-place
orphan-tail re-anchor was dead after a catch-up, so every retirement on a
dictionary-bearing slot cost a reconnect. The dead-arm half is right; the
cost claim is not. Both connection setup sites -- swapClient and
positionCursorForStart -- call tryRetireOrphanTail before any send, so
that arm is reached only when frames below the tail still needed acks,
which means they were sent on this connection and a recycle was always
the correct path. Proven: a test written for the fix passed with the fix
reverted. The code is restored; only the comment, which claimed the wire
mapping is untouched "because no wire sequence has been consumed" (false
once a catch-up has run), is corrected. The test is kept under a name
describing what it does pin -- that adding a catch-up leaves the
start()-time retirement intact -- and says so.

Full client suite: 2697 tests green.

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

@@ -155,7 +155,11 @@ public class QwpWebSocketSender implements Sender {
     private final QwpWebSocketEncoder encoder;
     private final List<Endpoint> endpoints;
     // Global symbol dictionary for delta encoding
-    private final GlobalSymbolDictionary globalSymbolDictionary;
+    // Not final: seedGlobalDictionaryFromPersisted replaces it with a pre-sized instance
+    // before anything can observe the original. Nothing retains a reference -- the encoder
+    // and the persisted dictionary both take it as a per-call parameter -- and the swap
+    // happens during construction, before the producer or the I/O thread exist.
+    private GlobalSymbolDictionary globalSymbolDictionary;
     // Serializes FOREGROUND connect walks only (see buildAndConnect): the
     // shared-round state in hostTracker (pickNext/beginRound/attempted
     // bits), roundSeq, roundConnectAttemptSeq, and the foreground lifecycle
@@ -4080,6 +4084,15 @@ private void seedGlobalDictionaryFromPersisted(PersistedSymbolDict pd) {
         // 1. The dictionary's intact prefix. addRecoveredSymbol appends without de-dup, so
         //    the producer's size tracks pd.size() exactly -- which is what the send loop's
         //    mirror also seeds sentDictCount from.
+        // Pre-size before pouring the recovered symbols in. The default capacity is 64,
+        // so rebuilding a large dictionary rehashed the map ~log2(n/64) times, each pass
+        // O(current size) -- roughly doubling the rebuild and touching a growing table
+        // the whole way. recoveredMaxSymbolId + 1 is the upper bound the seed can reach.
+        long expected = Math.max(pd == null ? 0L : pd.recoveredSize(),
+                cursorEngine.recoveredMaxSymbolId() + 1L);
+        if (expected > globalSymbolDictionary.size() && expected <= Integer.MAX_VALUE) {
+            globalSymbolDictionary = new GlobalSymbolDictionary((int) expected);
+        }
         int baseline = 0;
         if (pd != null && pd.size() > 0) {
             pd.addLoadedSymbolsTo(globalSymbolDictionary);

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

@@ -346,8 +346,6 @@ public final class CursorWebSocketSendLoop implements QuietCloseable {
     // connection in setWireBaselineWithCatchUp; set in trySendOne after a successful
     // send.
     private boolean dataFrameSentThisConnection;
-    // End position (native address) written by the last readVarintAt() call.
-    private long varintEnd;
     private WebSocketClient client;
     // Optional: when non-null, every server-rejection error (retriable and
     // terminal alike) is offered to the dispatcher for async delivery to the user's
@@ -2416,7 +2414,10 @@ private int frameDeltaStart(long payloadAddr, int payloadLen) {
         if (!isDeltaFrame(payloadAddr, payloadLen)) {
             return -1;
         }
-        return (int) readVarintAt(payloadAddr + QwpConstants.HEADER_SIZE, payloadAddr + payloadLen);
+        long encoded = readVarintAt(payloadAddr + QwpConstants.HEADER_SIZE, payloadAddr + payloadLen);
+        // A malformed start id reads as "no delta section", which the caller's
+        // `deltaStart >= 0` gate already handles.
+        return encoded < 0L ? -1 : (int) (encoded >>> 3);
     }
 
     // True only for a well-formed QWP frame this encoder produced that carries a
@@ -2453,8 +2454,12 @@ private void accumulateSentDict(long payloadAddr, int payloadLen, int deltaStart
         // its canonical LEB128 encoding rather than re-reading the header and the
         // start-id varint.
         long p = payloadAddr + QwpConstants.HEADER_SIZE + NativeBufferWriter.varintSize(deltaStart);
-        long deltaCount = readVarintAt(p, limit);
-        p = varintEnd;
+        long encodedCount = readVarintAt(p, limit);
+        if (encodedCount < 0L) {
+            return; // malformed -- never happens for frames we encoded
+        }
+        long deltaCount = encodedCount >>> 3;
+        p += encodedCount & 7L;
         // The mirror holds ids [0, sentDictCount). Accumulate ONLY the part of this
         // frame's delta [deltaStart, deltaStart+deltaCount) that extends past the
         // tip -- ids [sentDictCount, deltaStart+deltaCount). Cases:
@@ -2486,24 +2491,30 @@ private void accumulateSentDict(long payloadAddr, int payloadLen, int deltaStart
         // the new tail [sentDictCount, deltaEnd).
         int skip = sentDictCount - deltaStart;
         for (int i = 0; i < skip; i++) {
-            long len = readVarintAt(p, limit);
-            p = varintEnd + len;
-            if (p > limit) {
+            long encoded = readVarintAt(p, limit);
+            if (encoded < 0L) {
                 return; // malformed -- bail rather than corrupt the mirror
             }
+            p += (encoded & 7L) + (encoded >>> 3);
+            if (p > limit) {
+                return;
+            }
         }
         long regionStart = p;
         long newCount = deltaEnd - sentDictCount;
         // Walk the new tail only to find where it ends; the entry boundaries are not
         // recorded here (see above -- the catch-up re-walks the mirror when it needs them).
         for (long i = 0; i < newCount; i++) {
-            long len = readVarintAt(p, limit);
-            p = varintEnd + len;
-            if (p > limit) {
+            long encoded = readVarintAt(p, limit);
+            if (encoded < 0L) {
                 // Malformed -- never happens for frames we encoded; bail rather
                 // than corrupt the mirror.
                 return;
             }
+            p += (encoded & 7L) + (encoded >>> 3);
+            if (p > limit) {
+                return;
+            }
         }
         int regionBytes = (int) (p - regionStart);
         // long sum: sentDictBytesLen + regionBytes can exceed Integer.MAX_VALUE on
@@ -2569,28 +2580,34 @@ private void releaseSentDictBytes() {
         sentDictCount = 0;
     }
 
-    private long readVarintAt(long p, long limit) {
+    /**
+     * Decodes the varint at {@code [p, limit)} and returns {@code (value << 3) | bytes},
+     * or {@code -1} when it is truncated or runs past a canonical length.
+     * <p>
+     * Packing the consumed length into the return value -- the shape
+     * {@code RecoveredFrameAnalysis.readVarint} already uses -- drops a store to an
+     * instance field on every call, and this runs once per SYMBOL: per frame in
+     * accumulateSentDict, and once per mirror entry in sendDictCatchUp.
+     * <p>
+     * The {@code -1} also makes truncation DETECTABLE. The field version returned 0 with
+     * {@code varintEnd == p} when {@code p >= limit}, so a caller computing
+     * {@code p = varintEnd + len} got {@code p == limit} and its {@code p > limit}
+     * bail-out could not fire at an exact boundary -- guards that could not guard.
+     */
+    private static long readVarintAt(long p, long limit) {
         long value = 0;
         int shift = 0;
-        long cur = p;
-        while (cur < limit) {
-            byte b = Unsafe.getUnsafe().getByte(cur++);
+        int bytes = 0;
+        while (p < limit && bytes < 6) {
+            byte b = Unsafe.getUnsafe().getByte(p++);
             value |= (long) (b & 0x7F) << shift;
+            bytes++;
             if ((b & 0x80) == 0) {
-                break;
+                return (value << 3) | bytes;
             }
             shift += 7;
-            if (shift > 35) {
-                // Defensive bound, matching PersistedSymbolDict.decodeVarint: a
-                // canonical entry-length / delta varint is <= 5 bytes. Every caller
-                // reads freshly-encoded, CRC- or openExisting-validated bytes, so
-                // this is unreachable, but it stops a corrupt continuation run from
-                // over-shifting into a garbage length.
-                break;
-            }
         }
-        varintEnd = cur;
-        return value;
+        return -1L;
     }
 
     /**
@@ -2649,9 +2666,11 @@ private int sendDictCatchUp() {
         long mirrorLimit = sentDictBytesAddr + sentDictBytesLen;
         for (int entryId = 0; entryId < sentDictCount; entryId++) {
             long entryStart = entryPtr;
-            long entryLen = readVarintAt(entryPtr, mirrorLimit);
-            long entryEnd = varintEnd + entryLen;
-            if (entryEnd > mirrorLimit) {
+            long encodedLen = readVarintAt(entryPtr, mirrorLimit);
+            long entryEnd = encodedLen < 0L
+                    ? -1L
+                    : entryPtr + (encodedLen & 7L) + (encodedLen >>> 3);
+            if (entryEnd < 0L || entryEnd > mirrorLimit) {
                 LineSenderException err = new LineSenderException(
                         "invalid symbol dictionary mirror during catch-up [entry="
                                 + entryId + ", count=" + sentDictCount + ']');
@@ -3023,9 +3042,18 @@ private boolean trySendOne() {
                 return false;
             }
             if (nextWireSeq == 0) {
-                // Nothing sent on this connection yet: re-anchor in place past
-                // the retired tail. The wireSeq<->FSN mapping is untouched
-                // because no wire sequence has been consumed.
+                // Reached only when the tail was NOT retirable at connection setup --
+                // both setup sites (swapClient, positionCursorForStart) call
+                // tryRetireOrphanTail first -- which means frames below it still needed
+                // acks, which means they were SENT on this connection. So nextWireSeq is
+                // never 0 here in practice and this arm is effectively dead; it is kept
+                // as a cheap correctness guard rather than removed.
+                //
+                // NB the mapping is untouched only while NO wire sequence has been
+                // consumed. A dictionary catch-up consumes sequences 0..n-1 before any
+                // data frame, so after one this test is false and the recycle below --
+                // which re-anchors the mapping from scratch -- is the correct path, not
+                // an avoidable cost.
                 try {
                     positionCursorForStart();
                 } catch (CatchUpSendException e) {

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

@@ -933,7 +933,14 @@ private void ensureAppendMap(long required) {
                 && required <= appendMapOffset + appendMapCapacity) {
             return;
         }
-        long newOffset = appendOffset - appendOffset % APPEND_MAP_CAPACITY;
+        // Page-align, not APPEND_MAP_CAPACITY-align: mmap only requires a page-aligned
+        // file offset. Rounding the START down to a 4 MiB boundary while sizing `needed`
+        // to the record's END meant a chunk straddling that boundary produced a window
+        // spanning BOTH -- 8 MiB mapped and re-allocated, whose lower half covers bytes
+        // already written and never touched again. Steady state then advanced 4 MiB per
+        // remap while always mapping 8.
+        long pageMask = Files.PAGE_SIZE - 1;
+        long newOffset = appendOffset & ~pageMask;
         long needed = required - newOffset;
         long newCapacity = Math.max(APPEND_MAP_CAPACITY, needed);
         long remainder = newCapacity % APPEND_MAP_CAPACITY;

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

@@ -90,6 +90,7 @@ public final class SegmentRing implements QuietCloseable {
     private int sealedHead;
     // Deterministic performance probes: unlike elapsed-time assertions these
     // let tests pin binary successor search and amortized prefix compaction.
+    private int lastFindSegmentComparisons;
     private int lastNextSealedSearchComparisons;
     private long sealedCompactionMoves;
     // High-water byte offset within the active segment at which we proactively
@@ -608,8 +609,28 @@ public synchronized ObjList<MmapSegment> drainTrimmable() {
      * scan cost doesn't matter.
      */
     public synchronized MmapSegment findSegmentContaining(long fsn) {
-        for (int i = sealedHead, n = sealedSegments.size(); i < n; i++) {
-            MmapSegment s = sealedSegments.get(i);
+        // Binary search, for the same reason nextSealedAfter got one: the sealed list
+        // is sorted by baseSeq and its ranges are disjoint, and its length is bounded
+        // only by the documented ~16K-segment ceiling -- so the linear scan this
+        // replaces cost up to 16K comparisons under the ring monitor, on the reconnect
+        // path, right beside a sibling doing the identical predicate in log N.
+        int lo = sealedHead;
+        int hi = sealedSegments.size();
+        int comparisons = 0;
+        while (lo < hi) {
+            int mid = (lo + hi) >>> 1;
+            comparisons++;
+            if (sealedSegments.get(mid).baseSeq() <= fsn) {
+                lo = mid + 1;
+            } else {
+                hi = mid;
+            }
+        }
+        lastFindSegmentComparisons = comparisons;
+        // lo is the first segment whose baseSeq exceeds fsn, so the candidate is the
+        // one before it -- the only sealed segment whose range can contain fsn.
+        if (lo > sealedHead) {
+            MmapSegment s = sealedSegments.get(lo - 1);
             long base = s.baseSeq();
             if (fsn >= base && fsn < base + s.frameCount()) {
                 return s;
@@ -841,6 +862,11 @@ public synchronized long totalSegmentBytes() {
         return total;
     }
 
+    @TestOnly
+    public synchronized int getLastFindSegmentComparisons() {
+        return lastFindSegmentComparisons;
+    }
+
     @TestOnly
     public synchronized int getLastNextSealedSearchComparisons() {
         return lastNextSealedSearchComparisons;

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

@@ -184,6 +184,57 @@ public void testFastPathRetiresWholeDeferredLogBeforeAnySend() throws Exception
         });
     }
 
+    /**
+     * A whole-deferred slot that ALSO ships a dictionary catch-up must still retire its
+     * tail at start(), before any connection work, and must not reconnect.
+     * <p>
+     * testFastPathRetiresWholeDeferredLogBeforeAnySend covers the same shape without a
+     * dictionary, so nothing pinned that adding a catch-up -- which consumes wire
+     * sequences before any data frame -- leaves the start()-time retirement intact.
+     * <p>
+     * This deliberately does NOT pin trySendOne's in-place re-anchor arm: both connection
+     * setup sites call tryRetireOrphanTail first, so that arm is only reached when frames
+     * below the tail still needed acks -- which means they were sent on this connection.
+     * Verified by reverting its guard: this test still passes.
+     */
+    @Test
+    public void testCatchUpBearingWholeDeferredSlotRetiresAtStartWithoutReconnecting() throws Exception {
+        TestUtils.assertMemoryLeak(() -> {
+            try (CursorSendEngine engine = newEngine()) {
+                appendFrame(engine, true);
+                appendFrame(engine, true);
+                appendFrame(engine, true);
+            }
+            // A populated dictionary is what makes the loop ship a catch-up. The frames
+            // carry no symbols, so recoveredMaxSymbolId stays -1 and the full-dict
+            // discard cannot fire on it.
+            try (PersistedSymbolDict pd = PersistedSymbolDict.openClean(tmpDir)) {
+                assertNotNull(pd);
+                pd.appendSymbol("a");
+                pd.appendSymbol("b");
+            }
+
+            try (CursorSendEngine engine = newEngine()) {
+                assertEquals(-1L, engine.recoveredCommitBoundaryFsn());
+                assertEquals(2L, engine.recoveredOrphanTipFsn());
+
+                List<AckingClient> clients = new ArrayList<>();
+                CursorWebSocketSendLoop loop = newLoop(engine, clients);
+                try {
+                    loop.start();
+                    assertEquals("scaffolding: the mirror must be seeded, so a catch-up ships",
+                            2, loop.sentDictCount());
+                    awaitAckedFsn(engine, 2L);
+                    assertTrue("retiring after only a catch-up must re-anchor in place, "
+                                    + "not reconnect",
+                            clientCount(clients) <= 1);
+                } finally {
+                    loop.close();
+                }
+            }
+        });
+    }
+
     @Test
     public void testRecoveryReleasesAbortedTailSymbolStorage() throws Exception {
         TestUtils.assertMemoryLeak(() -> {

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

@@ -462,6 +462,25 @@ public void testNextSealedAfterWalksThousandsOfSegmentsWithoutOverflow() throws
                     // After the loop we have `sealedCount` sealed segments and one
                     // active (containing nothing yet — its base = sealedCount).
                     // Now walk: oldest sealed, then nextSealedAfter() repeatedly.
+                    // findSegmentContaining walks the SAME sorted, disjoint list and is
+                    // on the reconnect path, so it must be logarithmic too -- a linear
+                    // scan costs up to the ~16K-segment ceiling under the ring monitor.
+                    // The > 0 assertion is the anti-vacuity guard: the comparison counter
+                    // exists only in the binary search, so a revert to a linear scan would
+                    // leave it at 0 and make a bound-only assertion pass for free.
+                    int findCeiling = 33 - Integer.numberOfLeadingZeros(sealedCount);
+                    for (long probe = 0; probe < 16; probe++) {
+                        MmapSegment hit = ring.findSegmentContaining(probe);
+                        assertNotNull("fsn " + probe + " must resolve to a segment", hit);
+                        assertEquals("fsn " + probe + " lives in the segment based there",
+                                probe, hit.baseSeq());
+                        int comparisons = ring.getLastFindSegmentComparisons();
+                        assertTrue("the search must actually run", comparisons > 0);
+                        assertTrue("findSegmentContaining must be logarithmic: " + comparisons
+                                        + " comparisons over " + sealedCount + " sealed segments",
+                                comparisons <= findCeiling);
+                    }
+
                     MmapSegment cursor = ring.firstSealed();
                     assertNotNull(cursor);
                     assertEquals(0, cursor.baseSeq());