fix(qwp): drainer retries once over a chain the sealed-residue sanitize already healed

The fail-closed first-sight throw in sanitizeSealedResidue fires AFTER the
proven-dead sealed residue has been durably zeroed (msync+fsync), so the
chain on disk is already healed when it propagates. The orphan drainer
classified that throw as terminal SfRecoveryException and dropped a
.failed sentinel -- stranding a fully replayable backlog behind a
quarantine no scan revisits, for an incident recovery had just repaired.

Introduce SfSanitizedResidueException (a refinement of SfRecoveryException,
so producer startup keeps its fail-closed restart-proves-clean semantics)
and have the drainer intercept it for a single in-place construction
retry. Any second failure, including a repeat of the refinement from a
non-sticking heal, falls through to the existing terminal classification.

New drainer test proves the full arc: poisoned sealed suffix, heal durably
on disk before the throw, one retry, connect reached, no sentinel, backlog
still scanner-eligible. SegmentRingTest now pins the thrown type at the
sanitize site.

Author: bluestreak01

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

@@ -560,9 +560,29 @@ public void run() {
             // holds it, the engine constructor throws and we exit silently
             // (no .failed sentinel — contention is expected, not an error).
             try {
-                engine = new CursorSendEngine(slotPath, segmentSizeBytes,
-                        sfMaxTotalBytes, CursorSendEngine.DEFAULT_APPEND_DEADLINE_NANOS,
-                        syncIntervalNanos);
+                try {
+                    engine = new CursorSendEngine(slotPath, segmentSizeBytes,
+                            sfMaxTotalBytes, CursorSendEngine.DEFAULT_APPEND_DEADLINE_NANOS,
+                            syncIntervalNanos);
+                } catch (SfSanitizedResidueException first) {
+                    // First sight of proven-dead sealed residue: recovery
+                    // durably zeroed it BEFORE failing closed, so the chain
+                    // on disk is already healed and a .failed sentinel here
+                    // would strand a replayable backlog no scan revisits
+                    // (the sentinel gates isCandidateOrphan and nothing in
+                    // production clears it). Retry once over the healed
+                    // chain; the WARN keeps the incident surfaced. Any
+                    // failure of the retry is genuine and takes the normal
+                    // classification below -- including a repeat of this
+                    // type, which the SfRecoveryException arm then treats
+                    // as the terminal quarantine a non-sticking heal is.
+                    LOG.warn("drainer slot {}: sealed SF residue sanitized during recovery ({}); "
+                                    + "retrying engine construction over the healed chain",
+                            slotPath, first.getMessage());
+                    engine = new CursorSendEngine(slotPath, segmentSizeBytes,
+                            sfMaxTotalBytes, CursorSendEngine.DEFAULT_APPEND_DEADLINE_NANOS,
+                            syncIntervalNanos);
+                }
             } catch (SlotLockContentionException t) {
                 LOG.info("orphan slot already locked, skipping: {} ({})",
                         slotPath, t.getMessage());

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

@@ -36,7 +36,9 @@
  * <b>positively-identified corruption</b> in one file's own bytes, which
  * recovery may quarantine when the surviving chain proves safe; and
  * {@link SfRecoveryException} marks a <b>terminal chain failure</b> that needs
- * operator intervention.
+ * operator intervention (with one self-healed refinement,
+ * {@link SfSanitizedResidueException}, that unattended callers retry once
+ * instead of quarantining).
  */
 public class MmapSegmentException extends RuntimeException {
     public MmapSegmentException(String message) {

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

@@ -784,10 +784,13 @@ private static void validateContiguous(ObjList<MmapSegment> segments) {
      * fails recovery before any mutation, preserving the bytes (potentially
      * the only copy of unreachable valid-CRC frames) for operator
      * extraction. With {@code failClosedOnSight} the incident is still
-     * surfaced as a first-sight startup failure after sanitizing (the
-     * restart then proves the chain clean); without it the chain proceeds
-     * immediately (legacy migration, which predates the sealed-suffix
-     * contract).
+     * surfaced as a first-sight {@link SfSanitizedResidueException} after
+     * sanitizing: the residue is already durably zeroed when it propagates,
+     * so a retry proves the chain clean (attended callers get that via
+     * restart; unattended callers key off the distinct type to retry
+     * instead of quarantining a just-healed slot). Without the flag the
+     * chain proceeds immediately (legacy migration, which predates the
+     * sealed-suffix contract).
      */
     private static void sanitizeSealedResidue(ObjList<MmapSegment> chain, boolean failClosedOnSight) {
         String firstTornPath = null;
@@ -801,7 +804,7 @@ private static void sanitizeSealedResidue(ObjList<MmapSegment> chain, boolean fa
             }
         }
         if (failClosedOnSight && firstTornPath != null) {
-            throw new SfRecoveryException("corrupt torn tail in sealed SF segment " + firstTornPath);
+            throw new SfSanitizedResidueException("corrupt torn tail in sealed SF segment " + firstTornPath);
         }
     }
 

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

@@ -29,8 +29,14 @@
  * segment chain is corrupt or incomplete and requires operator intervention.
  * Operational filesystem failures continue to use {@link MmapSegmentException}
  * so callers can retry them without quarantining otherwise recoverable data.
+ * <p>
+ * One refinement is deliberately non-terminal for unattended callers:
+ * {@link SfSanitizedResidueException} marks a first-sight failure thrown
+ * AFTER recovery durably healed the chain, so a single retry validates
+ * clean. Catch sites that quarantine on this type must intercept the
+ * refinement first.
  */
-public final class SfRecoveryException extends MmapSegmentException {
+public class SfRecoveryException extends MmapSegmentException {
 
     public SfRecoveryException(String message) {
         super(message);

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

@@ -0,0 +1,50 @@
+/*+*****************************************************************************
+ *     ___                  _   ____  ____
+ *    / _ \ _   _  ___  ___| |_|  _ \| __ )
+ *   | | | | | | |/ _ \/ __| __| | | |  _ \
+ *   | |_| | |_| |  __/\__ \ |_| |_| | |_) |
+ *    \__\_\\__,_|\___||___/\__|____/|____/
+ *
+ *  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;
+
+/**
+ * First-sight failure over a chain that recovery has <b>already healed</b>.
+ * Thrown from the fail-closed branch of the sealed-residue sanitize: frame
+ * accounting was proven complete (contiguity plus boundary checks), the
+ * proven-dead suffix residue was durably zeroed ({@code msync} +
+ * {@code fsync} completed — a sync failure throws
+ * {@link MmapSegmentException} instead and never reaches this type), and
+ * the throw exists solely to surface the incident on the startup that
+ * observed it. An immediate re-open re-runs the same proofs over the
+ * zeroed suffix and succeeds.
+ * <p>
+ * Attended callers (producer startup, where an operator or supervisor
+ * restarts the process) should keep the parent's fail-closed semantics:
+ * the restart proves the chain clean. Unattended callers (the orphan
+ * drainer) may retry construction once instead of quarantining — dropping
+ * a {@code .failed} sentinel over a just-healed slot would strand its
+ * replayable backlog until an operator clears the sentinel by hand.
+ */
+public final class SfSanitizedResidueException extends SfRecoveryException {
+
+    public SfSanitizedResidueException(String message) {
+        super(message);
+    }
+}

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

@@ -27,6 +27,7 @@
 import io.questdb.client.cutlass.qwp.client.sf.cursor.AckWatermark;
 import io.questdb.client.cutlass.qwp.client.sf.cursor.BackgroundDrainer;
 import io.questdb.client.cutlass.qwp.client.sf.cursor.CursorSendEngine;
+import io.questdb.client.cutlass.qwp.client.sf.cursor.MmapSegment;
 import io.questdb.client.cutlass.qwp.client.sf.cursor.OrphanScanner;
 import io.questdb.client.std.Files;
 import io.questdb.client.std.MemoryTag;
@@ -196,6 +197,116 @@ public void testCorruptRecoveredChainIsQuarantined() throws Exception {
         });
     }
 
+    @Test
+    public void testSealedResidueFirstSightHealsAndDoesNotQuarantine() throws Exception {
+        TestUtils.assertMemoryLeak(() -> {
+            // A sealed member whose frame accounting is complete but whose
+            // suffix gap carries legacy pre-sanitization poison. Recovery
+            // durably zeroes the residue BEFORE failing closed on first
+            // sight (SegmentRing's sanitize-then-throw), so at the instant
+            // the drainer sees the throw the chain on disk is already
+            // healed. The drainer must retry construction over the healed
+            // chain and reach connect -- not strand the replayable backlog
+            // behind a .failed sentinel that no orphan scan revisits.
+            long segSize = MmapSegment.HEADER_SIZE
+                    + 4 * (MmapSegment.FRAME_HEADER_SIZE + 16)
+                    + 12; // sealed suffix gap that can never fit a frame
+            String p0Path = slotPath + "/p0.sfa";
+            String p1Path = slotPath + "/p1.sfa";
+            long gapStart;
+            long buf = Unsafe.malloc(16, MemoryTag.NATIVE_DEFAULT);
+            try {
+                Unsafe.getUnsafe().setMemory(buf, 16, (byte) 5);
+                MmapSegment s0 = MmapSegment.create(p0Path, 0, segSize);
+                for (int i = 0; i < 4; i++) {
+                    s0.tryAppend(buf, 16);
+                }
+                gapStart = s0.publishedOffset();
+                s0.close();
+                MmapSegment s1 = MmapSegment.create(p1Path, 4, segSize);
+                s1.tryAppend(buf, 16);
+                s1.close();
+            } finally {
+                Unsafe.free(buf, 16, MemoryTag.NATIVE_DEFAULT);
+            }
+
+            // Adopt the chain once so the slot carries a manifest, lock and
+            // watermark like any real producer slot (without a manifest the
+            // legacy-migration path sanitizes silently and never fails
+            // closed); close releases the lock.
+            try (CursorSendEngine seed = new CursorSendEngine(slotPath, segSize)) {
+                Assert.assertEquals("highest published FSN must cover frames 0..4",
+                        4L, seed.publishedFsn());
+            }
+
+            // Poison the sealed suffix gap the way a pre-sanitization
+            // client's reseal-after-recovery did.
+            int fd = Files.openRW(p0Path);
+            Assert.assertTrue("openRW must succeed", fd >= 0);
+            long junk = Unsafe.malloc(12, MemoryTag.NATIVE_DEFAULT);
+            try {
+                for (int i = 0; i < 3; i++) {
+                    Unsafe.getUnsafe().putInt(junk + i * 4L, 0xCAFEBABE);
+                }
+                Assert.assertEquals(12L, Files.write(fd, junk, 12, gapStart));
+                Files.fsync(fd);
+            } finally {
+                Unsafe.free(junk, 12, MemoryTag.NATIVE_DEFAULT);
+                Files.close(fd);
+            }
+
+            LinkageError injected = new LinkageError("injected drainer connect error");
+            AtomicInteger connectAttempts = new AtomicInteger();
+            BackgroundDrainer drainer = new BackgroundDrainer(
+                    slotPath,
+                    segSize,
+                    Long.MAX_VALUE,
+                    () -> {
+                        connectAttempts.incrementAndGet();
+                        throw injected;
+                    },
+                    5_000L,
+                    1L,
+                    10L,
+                    true,
+                    200L);
+
+            LinkageError thrown = null;
+            try {
+                drainer.run();
+            } catch (LinkageError e) {
+                thrown = e;
+            }
+
+            // The heal precedes the first-sight throw: frames intact,
+            // residue durably zeroed. This holds with or without the
+            // drainer fix -- it is exactly what makes quarantine wrong.
+            try (MmapSegment sealed = MmapSegment.openExisting(p0Path)) {
+                Assert.assertEquals("frames must be untouched", 4L, sealed.frameCount());
+                Assert.assertEquals("proven-dead residue must be zeroed on disk",
+                        0L, sealed.tornTailBytes());
+            }
+
+            Assert.assertSame("drainer must reach connect over the healed chain",
+                    injected, thrown);
+            Assert.assertEquals("exactly one connect attempt after the healed retry",
+                    1, connectAttempts.get());
+            TestUtils.assertContains(drainer.getLastErrorMessage(),
+                    "injected drainer connect error");
+            Assert.assertEquals(BackgroundDrainer.DrainOutcome.FAILED, drainer.outcome());
+            Assert.assertFalse("a just-healed slot must not be quarantined",
+                    Files.exists(slotPath + "/" + OrphanScanner.FAILED_SENTINEL_NAME));
+            Assert.assertTrue("replayable backlog must remain scanner-eligible",
+                    OrphanScanner.isCandidateOrphan(slotPath));
+            try (CursorSendEngine engine = new CursorSendEngine(slotPath, segSize)) {
+                Assert.assertEquals("backlog must remain fully replayable",
+                        4L, engine.publishedFsn());
+                Assert.assertTrue("drainer teardown must release the slot lock",
+                        OrphanScanner.isCandidateOrphan(slotPath));
+            }
+        });
+    }
+
     @Test
     public void testLockOpenFailureDoesNotQuarantineRecoverableData() throws Exception {
         TestUtils.assertMemoryLeak(() -> {

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

@@ -27,6 +27,7 @@
 import io.questdb.client.cutlass.qwp.client.sf.cursor.MmapSegment;
 import io.questdb.client.cutlass.qwp.client.sf.cursor.MmapSegmentException;
 import io.questdb.client.cutlass.qwp.client.sf.cursor.SegmentRing;
+import io.questdb.client.cutlass.qwp.client.sf.cursor.SfSanitizedResidueException;
 import io.questdb.client.std.Files;
 import io.questdb.client.std.MemoryTag;
 import io.questdb.client.std.Misc;
@@ -515,6 +516,9 @@ public void testProvenDeadSealedResidueSanitizedThenHealsAfterOneRestart() throw
                     Misc.free(SegmentRing.openExisting(tmpDir, segSize));
                     throw new AssertionError("poisoned sealed suffix must fail closed on first sight");
                 } catch (MmapSegmentException expected) {
+                    assertTrue("first-sight throw must be the healed-residue refinement so "
+                                    + "unattended callers can retry: " + expected.getClass().getName(),
+                            expected instanceof SfSanitizedResidueException);
                     assertTrue(expected.getMessage(),
                             expected.getMessage().contains("corrupt torn tail in sealed SF segment"));
                 }