Set aside the slots recovery cannot rescue, instead of failing

The full-dictionary rebuild and the unreplayable-slot quarantine answer
two halves of the same question and have to be ordered, not merged.

Rebuilding the dictionary from the surviving frames' own delta sections
rescues most of the slots that used to fail, so it must run FIRST -- a
quarantine that pre-judged the slot on its own coverage arithmetic would
set aside slots recovery can still replay in full, and strand data that
drains perfectly.

What it cannot rescue it still throws on, and that throw still escaped
build(). senderId is stable and a not-fully-drained slot is retained on
close, so every retry re-recovered the same slot and threw again: the
application could not construct a Sender at all, and so could not even
BUFFER new rows. One already-lost batch became an unbounded outage of
everything after it, which inverts the guarantee store-and-forward
exists to give.

So seedGlobalDictionaryFromPersisted is now the single authority on
whether a slot is replayable -- it is the only code that has tried every
source of truth -- and its give-up is typed. build() waits for that
verdict rather than forming its own, and on it sets the slot aside: the
bytes are kept for forensics and resend, the .failed sentinel tells the
orphan drainer the copy is human-in-the-loop, and the producer continues
on a clean slot. The engine's own isRecoveredDictionaryIncomplete()
guess goes away with it.

Not one frame of a set-aside slot ever reaches the wire; that safety
property is unchanged and still asserted.

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

Author: glasstiger

core/src/main/java/io/questdb/client/Sender.java

@@ -40,6 +40,7 @@
 import io.questdb.client.cutlass.qwp.client.sf.cursor.CursorWebSocketSendLoop;
 import io.questdb.client.cutlass.qwp.client.sf.cursor.OrphanScanner;
 import io.questdb.client.cutlass.qwp.client.sf.cursor.PersistedSymbolDict;
+import io.questdb.client.cutlass.qwp.client.sf.cursor.UnreplayableSlotException;
 import io.questdb.client.impl.ConfStringParser;
 import io.questdb.client.impl.ConfigString;
 import io.questdb.client.impl.ConfigView;
@@ -1560,11 +1561,6 @@ public Sender build() {
                 CursorSendEngine cursorEngine = new CursorSendEngine(
                         slotPath, actualSfMaxBytes,
                         actualSfMaxTotalBytes, actualSfAppendDeadlineNanos);
-                if (cursorEngine.isRecoveredDictionaryIncomplete()) {
-                    cursorEngine = quarantineTornSlot(
-                            cursorEngine, sfDir, senderId, slotPath, actualSfMaxBytes,
-                            actualSfMaxTotalBytes, actualSfAppendDeadlineNanos);
-                }
                 int actualErrorInboxCapacity = errorInboxCapacity != PARAMETER_NOT_SET_EXPLICITLY
                         ? errorInboxCapacity
                         : io.questdb.client.cutlass.qwp.client.sf.cursor.SenderErrorDispatcher.DEFAULT_CAPACITY;
@@ -1576,7 +1572,15 @@ public Sender build() {
                 for (int i = 0, n = hosts.size(); i < n; i++) {
                     wsEndpoints.add(new QwpWebSocketSender.Endpoint(hosts.getQuick(i), ports.getQuick(i)));
                 }
-                QwpWebSocketSender connected;
+                // The recovery seed inside connect() is the authority on whether a recovered
+                // slot can be replayed: it rebuilds the dictionary from its intact prefix and
+                // then from the surviving frames' own delta sections, and throws
+                // UnreplayableSlotException only once neither source holds the missing ids.
+                // Quarantining on anything weaker would set aside slots that recovery can
+                // still rescue, so build() waits for that verdict rather than pre-judging it.
+                QwpWebSocketSender connected = null;
+                boolean quarantined = false;
+                while (connected == null) {
                 try {
                     connected = QwpWebSocketSender.connect(
                             wsEndpoints,
@@ -1603,6 +1607,32 @@ public Sender build() {
                             actualPoisonMinEscalationWindowMillis,
                             actualCatchUpCapGapMinEscalationWindowMillis
                     );
+                } catch (UnreplayableSlotException e) {
+                    // The one failure build() recovers from. The slot's frames reference ids
+                    // that nothing still holds, so they can never go on the wire -- but that is
+                    // no reason to take the producer down with them. Before this, the throw
+                    // escaped build() and, because senderId is stable and a not-fully-drained
+                    // slot is retained on close, every retry re-recovered the same slot and
+                    // threw again: the application could not construct a Sender at all, so it
+                    // could not even BUFFER new rows. An already-lost batch became an unbounded
+                    // outage of everything after it.
+                    //
+                    // Set the slot aside instead, keep its bytes for forensics and resend, and
+                    // start the producer on a clean one. Once only: a second such failure would
+                    // mean the FRESH slot is unreplayable, which cannot happen, so let it out
+                    // rather than loop.
+                    if (quarantined || slotPath == null) {
+                        try {
+                            cursorEngine.close();
+                        } catch (Throwable ignored) {
+                            // best-effort
+                        }
+                        throw e;
+                    }
+                    quarantined = true;
+                    cursorEngine = quarantineTornSlot(
+                            cursorEngine, e, sfDir, senderId, slotPath, actualSfMaxBytes,
+                            actualSfMaxTotalBytes, actualSfAppendDeadlineNanos);
                 } catch (Throwable t) {
                     // connect() failed before ownership of cursorEngine
                     // transferred — close it ourselves.
@@ -1613,6 +1643,7 @@ public Sender build() {
                     }
                     throw t;
                 }
+                }
                 // connect() succeeded — `connected` now owns cursorEngine
                 // via setCursorEngine(engine, true). From here on, ANY
                 // failure must close `connected` (which closes the engine
@@ -3001,18 +3032,17 @@ private static long parseSizeValue(@NotNull StringSink value, @NotNull String na
          * throw -- loudly, and never silently dropping bytes.
          */
         private static CursorSendEngine quarantineTornSlot(
-                CursorSendEngine torn, String sfDir, String senderId, String slotPath,
+                CursorSendEngine torn, UnreplayableSlotException cause, String sfDir,
+                String senderId, String slotPath,
                 long sfMaxBytes, long sfMaxTotalBytes, long sfAppendDeadlineNanos
         ) {
-            long recoveredMaxSymbolId = torn.recoveredMaxSymbolId();
-            PersistedSymbolDict dict = torn.getPersistedSymbolDict();
-            int coverage = dict != null ? dict.size() : 0;
-            String detail = "recovered store-and-forward symbol dictionary cannot cover the surviving "
-                    + "frames (likely a host crash tore its unsynced tail): frames reference symbol id "
-                    + recoveredMaxSymbolId + " but the recovered dictionary holds only " + coverage
-                    + " id(s)";
-            // Release the slot lock and the dictionary fd before renaming. The slot is not
-            // fully drained, so close() retains every file.
+            // The verdict, and the reason, come from the recovery seed -- the only code that
+            // has tried every source of truth. Recomputing them here would mean a second,
+            // independently-drifting notion of "unreplayable".
+            final String detail = cause.getMessage();
+            // Release the slot lock and the dictionary fd before renaming. connect()'s failure
+            // path already closed the engine; close() is idempotent, so make it explicit rather
+            // than depend on that.
             torn.close();
 
             String quarantinePath = null;

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

@@ -47,6 +47,7 @@
 import io.questdb.client.cutlass.qwp.client.sf.cursor.DefaultSenderErrorHandler;
 import io.questdb.client.cutlass.qwp.client.sf.cursor.DefaultSenderProgressHandler;
 import io.questdb.client.cutlass.qwp.client.sf.cursor.PersistedSymbolDict;
+import io.questdb.client.cutlass.qwp.client.sf.cursor.UnreplayableSlotException;
 import io.questdb.client.cutlass.qwp.client.sf.cursor.SenderConnectionDispatcher;
 import io.questdb.client.cutlass.qwp.client.sf.cursor.SenderErrorDispatcher;
 import io.questdb.client.cutlass.qwp.client.sf.cursor.SenderProgressDispatcher;
@@ -3926,7 +3927,10 @@ private void seedGlobalDictionaryFromPersisted(PersistedSymbolDict pd) {
         ObjList<String> fromFrames = new ObjList<>();
         long coverage = cursorEngine.collectReplaySymbolsAbove(baseline, fromFrames);
         if (coverage < 0) {
-            throw new LineSenderException(
+            // Typed, because Sender.build() sets such a slot aside instead of failing: this
+            // is the point at which every source of truth has been tried and none of them
+            // holds the missing ids. See UnreplayableSlotException.
+            throw new UnreplayableSlotException(
                     "recovered store-and-forward symbol dictionary is incomplete and cannot be "
                             + "rebuilt from the surviving frames (likely a host crash tore its unsynced "
                             + "tail): the frames reference symbol ids below their own delta start, which "

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

@@ -767,37 +767,6 @@ public boolean isDeltaDictEnabled() {
         return sfDir == null || persistedSymbolDict != null;
     }
 
-    /**
-     * Whether this RECOVERED slot's surviving frames reference symbol ids the
-     * recovered dictionary cannot cover -- i.e. the slot cannot be drained by a
-     * producer that shares its id space, and a producer seeded from the short
-     * dictionary would reuse ids those frames already define.
-     * <p>
-     * True in two situations, both of which mean the same thing to the caller:
-     * <ul>
-     *   <li>the dictionary opened but is SHORT of the frames (a host/power crash
-     *       tore its unsynced tail -- see {@code PersistedSymbolDict}'s durability
-     *       note); or</li>
-     *   <li>the dictionary did not open at all ({@code persistedSymbolDict == null}
-     *       -- an unreadable or torn side-file, fd exhaustion, a read-only remount)
-     *       while the surviving frames are DELTA frames that need one. Coverage is
-     *       then zero, so any frame referencing an id at all is unreplayable.</li>
-     * </ul>
-     * Always false for a fresh (non-recovered) slot: {@code recoveredMaxSymbolId} is
-     * -1 there, and -1 is below every coverage.
-     * <p>
-     * The caller must NOT drain such a slot with a live producer attached. The
-     * foreground sender quarantines it and starts on a fresh slot
-     * ({@code Sender.build()}); the orphan drainer reaches the same verdict
-     * independently via the send loop's replay guard and marks the slot failed.
-     * Either way the recorded bytes stay on disk and the affected data must be
-     * resent -- but the producer keeps producing, which is the whole point of
-     * store-and-forward.
-     */
-    public boolean isRecoveredDictionaryIncomplete() {
-        long coverage = persistedSymbolDict != null ? persistedSymbolDict.size() : 0;
-        return recoveredMaxSymbolId >= coverage;
-    }
 
     /**
      * Pass-through to {@link SegmentRing#nextSealedAfter(MmapSegment)}.

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

@@ -0,0 +1,51 @@
+/*******************************************************************************
+ *     ___                  _   ____  ____
+ *    / _ \ _   _  ___  ___| |_|  _ \| __ )
+ *   | | | | | | |/ _ \/ __| __| | | |  _ \
+ *   | |_| | |_| |  __/\__ \ |_| |_| | |_) |
+ *    \__\_\\__,_|\___||___/\__|____/|____/
+ *
+ *  Copyright (c) 2014-2019 Appsicle
+ *  Copyright (c) 2019-2026 QuestDB
+ *
+ *  Licensed under the Apache License, Version 2.0 (the "License");
+ *  you may not use this file except in compliance with the License.
+ *  You may obtain a copy of the License at
+ *
+ *  http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ *
+ ******************************************************************************/
+
+package io.questdb.client.cutlass.qwp.client.sf.cursor;
+
+import io.questdb.client.cutlass.line.LineSenderException;
+
+/**
+ * A recovered store-and-forward slot whose symbol dictionary cannot be rebuilt at all --
+ * neither from the persisted side-file's intact prefix nor from the surviving frames' own
+ * delta sections. Its frames reference ids that nothing still holds, so replaying them
+ * would null-pad the gap on the server and silently misattribute symbol values.
+ * <p>
+ * This is the ONE verdict {@code seedGlobalDictionaryFromPersisted} reaches after it has
+ * tried every source of truth it has, so it is the only signal that may quarantine a slot.
+ * A distinct type, rather than a message match, because the difference between "this slot
+ * is unreplayable" and any other {@link LineSenderException} out of the connect path is the
+ * difference between setting a slot aside and silently discarding one that was fine.
+ * <p>
+ * It stays a {@code LineSenderException} so that a caller which does NOT handle it -- the
+ * background drainer, a test constructing the sender directly -- keeps the old fail-clean
+ * behaviour rather than seeing a new checked type. Only {@code Sender.build()} treats it as
+ * recoverable, by setting the slot aside and starting the producer on a fresh one.
+ */
+public class UnreplayableSlotException extends LineSenderException {
+
+    public UnreplayableSlotException(CharSequence message) {
+        super(message);
+    }
+}