Close store-and-forward recovery and drainer gaps

Four fixes across the recovery, orphan-scan and drainer paths, all on
inputs a real cluster can produce.

Bound the oversized-batch throw. flushPendingRowsSplit's all-or-nothing
pre-flight throws when a single table's frame exceeds the server cap, and
keeps the batch buffered so a rejected flush does not silently drop the
caller's rows -- correct, and better than the old behaviour that discarded
every table's rows while reporting failure. But a batch that can never fit
the reachable cap then re-rejects on every flush(), growing unbounded if
the caller keeps appending. The batch cannot drain until a larger-cap node
returns or the sender is closed, so the message now says so, rather than
reading as a transient a caller would retry into an ever-larger batch.

Stop OrphanScanner counting the legacy-reader guards as data.
CursorSendEngine plants .qwp-v2-guard-a/b.sfa in every disk-mode slot,
before recovery or any segment, and its failure path does not unlink them.
A construction that then throws (ENOSPC on the initial segment) leaves a
slot holding only guards, which the scanner reported as an orphan with
unacked data: a drainer adopted it, failed its own build under the same
disk pressure, and dropped a permanent .failed sentinel plus an ERROR on a
slot that never held a byte. Exclude the guards by name (SegmentRing's own
predicate, now shared) -- a real segment beside them still counts.

Saturate the drainer's capability-gap budget. reconnect_max_duration_millis
is validated only as > 0, so a large value (Long.MAX_VALUE is the natural
"never give up") wrapped the raw millis*1_000_000 multiply negative, and
the OR-gated elapsed check then quarantined the slot on the FIRST
capability-gap sweep, skipping the whole 16-sweep settle budget -- asking
for more tolerance bought none. Convert with TimeUnit.toNanos, which
clamps at Long.MAX_VALUE. The keepalive-interval and poison-escalation
windows in CursorWebSocketSendLoop's constructor had the identical raw
multiply -- the poison one is the exact twin of the cap-gap dwell the PR
already hardened -- and a negative there escalates a transient to a
producer-fatal terminal on strike count alone; convert both the same way.

Make the legacy-reader barrier durable and stop re-truncating it. Its
content never changes, yet it was rewritten via openCleanRW (truncate) on
every engine open and never synced, so every open re-opened a window in
which a host crash leaves a zero-filled guard beside already-durable v2
segments -- and a rolled-back v1 reader then skips the guard (bad magic)
and the segments (bad version) alike, reads the slot empty, and truncates
the unacked log. Verify an existing guard and rewrite only a missing or
damaged one, and msync the guard on creation (33 bytes, once per slot
lifetime) so it is durable before the data it protects. The guard frame is
not byte-reproducible, so the test discriminates kept from rewritten by
byte-identity across a reopen and asserts only semantic validity after a
repair.

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

@@ -3702,10 +3702,23 @@ private void flushPendingRowsSplit(
                     splitFrameBodyBytes.getQuick(bodyIdx), simBaseline, currentBatchMaxSymbolId);
             bodyIdx++;
             if (messageSize > cap) {
+                // The batch stays BUFFERED: this throw precedes every publish, and a
+                // rejected flush must not silently discard the caller's rows (see
+                // SelfSufficientFramesTest#testOversizedTableSplitStrandsNothing). So the
+                // next flush() re-encodes and re-rejects the same batch until either the
+                // reachable cap grows -- a failover to a larger-cap node, which is the
+                // case retaining the rows exists to survive -- or the sender is closed,
+                // which discards them. It cannot drain against a cap this table will
+                // never fit, so say that here rather than let a caller read the repeat
+                // rejections as a transient and keep appending to a batch that only
+                // grows.
                 throw new LineSenderException("single table batch too large for server batch cap")
                         .put(" [table=").put(tableName)
                         .put(", messageSize=").put(messageSize)
-                        .put(", serverMaxBatchSize=").put(cap).put(']');
+                        .put(", serverMaxBatchSize=").put(cap).put(']')
+                        .put("; the batch is retained for retry and every flush() will "
+                                + "reject it again until a larger-cap node is reached -- "
+                                + "close the sender to discard it, or produce smaller batches");
             }
             // Mirror advanceSentMaxSymbolId: once the first frame ships the batch's
             // new ids, the remaining frames carry an empty delta above the baseline.

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

@@ -36,6 +36,7 @@
 import org.slf4j.LoggerFactory;
 
 import java.util.concurrent.ThreadLocalRandom;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.locks.LockSupport;
 
 /**
@@ -287,7 +288,17 @@ public WebSocketClient connectWithDurableAckRetry() {
         // Timestamp of the previous capability-gap sweep; 0 = the next gap
         // charges nothing because a fresh episode is starting.
         long lastCapabilityGapNanos = 0L;
-        final long capabilityGapBudgetNanos = reconnectMaxDurationMillis * 1_000_000L;
+        // Saturate rather than multiply: reconnect_max_duration_millis is validated only
+        // as > 0, so a large value (Long.MAX_VALUE is the natural way to ask for "never
+        // give up") wraps a raw multiply NEGATIVE. capabilityGapElapsedNanos then clears
+        // the budget on the FIRST capability-gap sweep -- 0 >= a negative -- and, because
+        // this gate is an OR with the attempt cap, nothing else holds it back: the slot
+        // quarantines immediately, skipping the whole 16-sweep settle budget. Asking for
+        // more tolerance would buy exactly none. TimeUnit clamps at Long.MAX_VALUE, which
+        // is the intended "effectively unbounded". CursorWebSocketSendLoop's dwell
+        // conversion guards the same way for the same reason.
+        final long capabilityGapBudgetNanos =
+                TimeUnit.MILLISECONDS.toNanos(reconnectMaxDurationMillis);
         // Observability-only counter for the transient all-replica window;
         // never consulted for escalation (Invariant B).
         int roleRejectAttempts = 0;

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

@@ -839,12 +839,20 @@ public CursorWebSocketSendLoop(WebSocketClient client, CursorSendEngine engine,
         this.reconnectInitialBackoffMillis = reconnectInitialBackoffMillis;
         this.reconnectMaxBackoffMillis = reconnectMaxBackoffMillis;
         this.durableAckMode = durableAckMode;
+        // Saturate, never multiply raw -- the same hazard the cap-gap dwell above
+        // guards. A raw multiply wraps a large millisecond value NEGATIVE, and both of
+        // these read as "elapsed >= window", so a negative makes the gate trivially
+        // true: the keepalive would ping on every loop pass, and the poison detector
+        // would escalate on the strike count alone -- exactly the transient-to-terminal
+        // false positive the dwell exists to stop (Invariant B). Both keys are
+        // user-supplied and validated only as >= 0, so asking for a very long window is
+        // legal, and asking for a longer one must never buy a shorter one.
         this.durableAckKeepaliveIntervalNanos = durableAckKeepaliveIntervalMillis > 0
-                ? durableAckKeepaliveIntervalMillis * 1_000_000L
+                ? TimeUnit.MILLISECONDS.toNanos(durableAckKeepaliveIntervalMillis)
                 : 0L;
         this.maxHeadFrameRejections = maxHeadFrameRejections;
         this.poisonMinEscalationWindowNanos = poisonMinEscalationWindowMillis > 0
-                ? poisonMinEscalationWindowMillis * 1_000_000L
+                ? TimeUnit.MILLISECONDS.toNanos(poisonMinEscalationWindowMillis)
                 : 0L;
         // SYNC/OFF startup hands a live client to the constructor, so we
         // already know we reached the server at least once. ASYNC startup

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

@@ -194,6 +194,15 @@ static void createLegacyReaderGuard(String path, long baseSeq) {
             if (segment.tryAppend(payload, 1) != HEADER_SIZE) {
                 throw new MmapSegmentException("could not append legacy-reader guard frame: " + path);
             }
+            // Unlike the segment log this file guards, the guard IS msync'd. It is a
+            // rollback barrier, so it has to be durable before the data it protects
+            // becomes durable -- a guard still sitting in the page cache when the host
+            // dies leaves zeroes on disk beside written-back v2 segments, and a
+            // rolled-back v1 reader skips the guard (bad magic) and the segments (bad
+            // version) alike, sees an empty slot, and truncates the log. It is 33 bytes
+            // written once per slot lifetime, not per flush, so the sync costs nothing
+            // measurable and buys the barrier the durability its whole purpose assumes.
+            segment.msync();
         } finally {
             if (payload != 0L) {
                 Unsafe.free(payload, 1, MemoryTag.NATIVE_DEFAULT);

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

@@ -327,7 +327,17 @@ private static boolean hasAnySegmentFile(String slotPath) {
             while (rc > 0) {
                 String name = Files.utf8ToString(Files.findName(find));
                 rc = Files.findNext(find);
-                if (name != null && name.endsWith(".sfa")) {
+                // The legacy-reader barriers are named .sfa on purpose (a rolled-back v1
+                // reader must not skip them), but they are a barrier, not data. Counting
+                // them here would call a slot that never held a byte an orphan with
+                // unacked data: CursorSendEngine plants them FIRST, before recovery or
+                // any segment is created, and its cleanup does not unlink them, so a
+                // construction that fails afterwards (ENOSPC on the initial segment)
+                // leaves a directory holding nothing else. A drainer would then adopt it,
+                // fail its own build under the same disk pressure, and quarantine an
+                // empty slot with a permanent .failed sentinel plus an ERROR for an
+                // operator to chase.
+                if (name != null && name.endsWith(".sfa") && !SegmentRing.isLegacyReaderGuard(name)) {
                     return true;
                 }
             }

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

@@ -349,12 +349,47 @@ public long ackedFsn() {
      * logical ring.
      */
     static void installLegacyReaderBarrier(String sfDir) {
-        MmapSegment.createLegacyReaderGuard(
-                sfDir + '/' + LEGACY_READER_GUARD_A,
-                Long.MAX_VALUE - 4L);
-        MmapSegment.createLegacyReaderGuard(
-                sfDir + '/' + LEGACY_READER_GUARD_B,
-                Long.MAX_VALUE - 1L);
+        installLegacyReaderGuardIfDamaged(sfDir + '/' + LEGACY_READER_GUARD_A, Long.MAX_VALUE - 4L);
+        installLegacyReaderGuardIfDamaged(sfDir + '/' + LEGACY_READER_GUARD_B, Long.MAX_VALUE - 1L);
+    }
+
+    /**
+     * Verifies the guard at {@code path} and rewrites it only when it is missing or
+     * damaged.
+     * <p>
+     * A guard's content never changes, so re-creating one that is already intact buys
+     * nothing -- and costs the only window in which the barrier can be lost. Creation
+     * goes through {@code openCleanRW} (which truncates) and does not fsync, so an
+     * unconditional rewrite on every engine open re-opens that window every time. A host
+     * crash inside it leaves a zero-filled guard next to v2 segment data that is already
+     * durable; a rolled-back v1 reader then skips the guard (bad magic) AND the v2
+     * segments (bad version), concludes the slot is empty, and truncates the unacked log
+     * the barrier exists to protect. Keeping an intact guard means the steady state never
+     * enters that window at all.
+     */
+    private static void installLegacyReaderGuardIfDamaged(String path, long baseSeq) {
+        if (isIntactLegacyReaderGuard(path, baseSeq)) {
+            return;
+        }
+        MmapSegment.createLegacyReaderGuard(path, baseSeq);
+    }
+
+    private static boolean isIntactLegacyReaderGuard(String path, long baseSeq) {
+        if (!Files.exists(path)) {
+            return false;
+        }
+        try (MmapSegment guard = MmapSegment.openExisting(path)) {
+            // Anything short of the exact shape createLegacyReaderGuard writes -- a
+            // zero-filled survivor of a torn creation included -- is rewritten.
+            return guard != null
+                    && guard.version() == MmapSegment.LEGACY_VERSION
+                    && guard.baseSeq() == baseSeq
+                    && guard.frameCount() == 1
+                    && guard.tornTailBytes() == 0L;
+        } catch (Throwable t) {
+            LOG.warn("legacy-reader guard {} could not be verified ({}); rewriting it", path, t.toString());
+            return false;
+        }
     }
 
     /**
@@ -945,7 +980,15 @@ private static void sortByBaseSeq(ObjList<MmapSegment> list, int lo, int hi) {
         }
     }
 
-    private static boolean isLegacyReaderGuard(String name) {
+    /**
+     * Whether {@code name} is one of the reserved rollback-barrier files this class
+     * plants in every disk-mode slot. They carry the {@code .sfa} suffix so a rolled-back
+     * v1 reader cannot skip them, which means every directory walk that reasons about
+     * "does this slot hold data" has to exclude them -- they are a barrier, not a log.
+     * Package-private so {@link OrphanScanner} shares this one definition rather than
+     * matching the names itself.
+     */
+    static boolean isLegacyReaderGuard(String name) {
         return LEGACY_READER_GUARD_A.equals(name) || LEGACY_READER_GUARD_B.equals(name);
     }
 

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

@@ -47,6 +47,7 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.locks.LockSupport;
 
+import static org.junit.Assert.assertArrayEquals;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNotNull;
@@ -298,6 +299,62 @@ public void testRecoveryAdvancesAckedFsnPastWatermark() throws Exception {
         });
     }
 
+    @Test
+    public void testLegacyReaderGuardsSurviveReopenAndAreRepairedWhenDamaged() throws Exception {
+        // The barrier is only worth what it is worth on disk. Creating it truncates
+        // (openCleanRW) and, before this, did not sync -- and it ran unconditionally on
+        // EVERY engine open, so every open destroyed a durable guard and re-dirtied it.
+        // A host crash inside that writeback window leaves a zero-filled guard beside v2
+        // segments that ARE already durable; a rolled-back v1 reader then skips the guard
+        // (bad magic) and the segments (bad version) alike, reads the slot as empty, and
+        // truncates the unacked log the guard was planted to protect. The guard's content
+        // never changes, so an intact one is verified and kept, and only a damaged one is
+        // rewritten -- the steady state never re-enters the window.
+        //
+        // Two guards created with identical arguments are NOT byte-identical (they
+        // differ in the frame header region), so a KEPT guard is byte-identical to
+        // itself across a reopen while a REWRITTEN one is not -- which is exactly the
+        // discriminator this needs. The repair leg then only asserts semantic validity,
+        // since a rebuilt guard legitimately differs byte-for-byte from the original.
+        TestUtils.assertMemoryLeak(() -> {
+            long buf = Unsafe.malloc(16, MemoryTag.NATIVE_DEFAULT);
+            try {
+                try (CursorSendEngine engine = new CursorSendEngine(tmpDir, 4096)) {
+                    assertEquals(0L, engine.appendBlocking(buf, 16));
+                }
+                String guardA = tmpDir + "/.qwp-v2-guard-a.sfa";
+                byte[] planted = java.nio.file.Files.readAllBytes(java.nio.file.Paths.get(guardA));
+                assertTrue("the guard must carry bytes", planted.length > 0);
+
+                // Reopening the slot must leave an intact guard untouched. Under the old
+                // unconditional re-create this rewrites it, and because the frame is not
+                // byte-reproducible the bytes then differ -- so this comparison fails.
+                try (CursorSendEngine reopened = new CursorSendEngine(tmpDir, 4096)) {
+                    assertEquals("recovery must adopt the one real frame, not a guard",
+                            0L, reopened.publishedFsn());
+                }
+                assertArrayEquals("an intact guard must survive a reopen untouched",
+                        planted, java.nio.file.Files.readAllBytes(java.nio.file.Paths.get(guardA)));
+
+                // Zero it -- the shape a torn creation leaves -- and it must be rebuilt to
+                // a valid guard, otherwise a slot that crashed mid-plant stays unprotected.
+                java.nio.file.Files.write(java.nio.file.Paths.get(guardA), new byte[planted.length]);
+                try (CursorSendEngine repaired = new CursorSendEngine(tmpDir, 4096)) {
+                    assertEquals("recovery must adopt the one real frame, not a guard",
+                            0L, repaired.publishedFsn());
+                }
+                try (MmapSegment guard = MmapSegment.openExisting(guardA)) {
+                    assertNotNull("a damaged guard must be rewritten to a valid one", guard);
+                    assertEquals(MmapSegment.LEGACY_VERSION, guard.version());
+                    assertEquals(1L, guard.frameCount());
+                    assertEquals("a rewritten guard must not read as torn", 0L, guard.tornTailBytes());
+                }
+            } finally {
+                Unsafe.free(buf, 16, MemoryTag.NATIVE_DEFAULT);
+            }
+        });
+    }
+
     @Test
     public void testV2DiskSlotForcesLegacyReaderToFailClosed() throws Exception {
         TestUtils.assertMemoryLeak(() -> {

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

@@ -89,6 +89,34 @@ public void testSlotWithSfaIsAnOrphan() throws Exception {
         });
     }
 
+    @Test
+    public void testSlotHoldingOnlyLegacyReaderGuardsIsNotAnOrphan() throws Exception {
+        TestUtils.assertMemoryLeak(() -> {
+            // The rollback barriers are named .sfa deliberately -- a rolled-back v1
+            // reader must not skip them -- but they are a barrier, not data. A slot
+            // holding nothing else has never had a byte written to it and must not be
+            // adopted: CursorSendEngine plants the guards FIRST, before recovery or the
+            // initial segment, and its failure path does not unlink them, so an ENOSPC
+            // partway through construction leaves exactly this directory. Counting the
+            // guards sent a drainer to adopt it, which fails under the same disk
+            // pressure and quarantines the empty slot with a permanent .failed sentinel.
+            String slot = sfDir + "/guards-only";
+            assertEquals(0, Files.mkdir(slot, Files.DIR_MODE_DEFAULT));
+            touchFile(slot + "/.qwp-v2-guard-a.sfa");
+            touchFile(slot + "/.qwp-v2-guard-b.sfa");
+
+            ObjList<String> orphans = OrphanScanner.scan(sfDir, "default");
+            assertEquals(0, orphans.size());
+
+            // ...but a real segment alongside them still is: the exclusion must be by
+            // guard name, not "ignore dot-prefixed .sfa" or "ignore the first two".
+            touchFile(slot + "/sf-0001.sfa");
+            orphans = OrphanScanner.scan(sfDir, "default");
+            assertEquals(1, orphans.size());
+            assertEquals(slot, orphans.get(0));
+        });
+    }
+
     @Test
     public void testEmptySlotDirIsNotAnOrphan() throws Exception {
         TestUtils.assertMemoryLeak(() -> {