fix(qwp): harden orphan recovery and metadata crash safety

Author: bluestreak01

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

@@ -39,6 +39,7 @@
 import io.questdb.client.cutlass.line.LineSenderException;
 import io.questdb.client.cutlass.line.array.DoubleArray;
 import io.questdb.client.cutlass.line.array.LongArray;
+import io.questdb.client.cutlass.qwp.client.sf.cursor.BackgroundDrainer;
 import io.questdb.client.cutlass.qwp.client.sf.cursor.BackgroundDrainerListener;
 import io.questdb.client.cutlass.qwp.client.sf.cursor.BackgroundDrainerPool;
 import io.questdb.client.cutlass.qwp.client.sf.cursor.CursorSendEngine;
@@ -2330,7 +2331,7 @@ public synchronized void setDrainerListener(BackgroundDrainerListener listener)
             pool.setListener(listener);
             // ...and direct re-assignment for the ones already running (the
             // pool listener is only applied at submit time, never after).
-            ObjList<io.questdb.client.cutlass.qwp.client.sf.cursor.BackgroundDrainer> live =
+            ObjList<BackgroundDrainer> live =
                     pool.snapshot();
             for (int i = 0, n = live.size(); i < n; i++) {
                 live.getQuick(i).setListener(listener);
@@ -2473,16 +2474,16 @@ public synchronized void startOrphanDrainers(
             // (the factory needs the drainer, the drainer's constructor
             // needs the factory); the ref write happens-before the drainer
             // runs because submit() publishes the task afterwards.
-            final io.questdb.client.cutlass.qwp.client.sf.cursor.BackgroundDrainer[] ref =
-                    new io.questdb.client.cutlass.qwp.client.sf.cursor.BackgroundDrainer[1];
+            final BackgroundDrainer[] ref =
+                    new BackgroundDrainer[1];
             ReconnectSupplier factory = new ReconnectSupplier(
                     () -> {
-                        io.questdb.client.cutlass.qwp.client.sf.cursor.BackgroundDrainer d = ref[0];
+                        BackgroundDrainer d = ref[0];
                         return d != null && d.isStopRequested();
                     },
                     "drainer stop requested during connect");
-            io.questdb.client.cutlass.qwp.client.sf.cursor.BackgroundDrainer drainer =
-                    new io.questdb.client.cutlass.qwp.client.sf.cursor.BackgroundDrainer(
+            BackgroundDrainer drainer =
+                    new BackgroundDrainer(
                             slot, segmentSizeBytes, sfMaxTotalBytes,
                             syncIntervalNanos,
                             factory,

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

@@ -45,8 +45,10 @@
  * "everything <= N is durable"), so a single monotonic watermark suffices;
  * no per-frame bitmap is needed.
  * <p>
- * <b>Layout</b> (128 bytes, little-endian, mmap'd for the engine's lifetime):
- * two independently CRC-protected 64-byte records. Each record contains:
+ * <b>Layout</b> (8192 bytes, little-endian, mmap'd for the engine's lifetime):
+ * two independently CRC-protected 64-byte records at offsets 0 and 4096.
+ * Placing each record at the start of a separate 4 KiB slot prevents one
+ * aligned 512-byte or 4 KiB sector tear from damaging both. Each record contains:
  * <pre>
  *   offset 0:   u32 magic = 'AKW1'
  *   offset 4:   u32 version = 1
@@ -84,7 +86,7 @@ public final class AckWatermark implements QuietCloseable {
      * directory enumerators that filter by extension skip it automatically.
      */
     public static final String FILE_NAME = ".ack-watermark";
-    public static final int FILE_SIZE = 128;
+    public static final int FILE_SIZE = 8 * 1024;
     /**
      * Sentinel returned by {@link #read()} when neither watermark record is
      * valid.
@@ -96,6 +98,7 @@ public final class AckWatermark implements QuietCloseable {
     private static final Logger LOG = LoggerFactory.getLogger(AckWatermark.class);
     private static final int MAGIC_OFFSET = 0;
     private static final int RECORD_SIZE = 64;
+    private static final int RECORD_SLOT_SIZE = 4 * 1024;
     private static final int VERSION = 1;
     private final int fd;
     private final FilesFacade filesFacade;
@@ -240,7 +243,7 @@ public void sync() {
     public void write(long fsn) {
         if (closed) return;
         long nextGeneration = generation + 1L;
-        long recordAddress = mmapAddress + (nextGeneration & 1L) * RECORD_SIZE;
+        long recordAddress = mmapAddress + (nextGeneration & 1L) * RECORD_SLOT_SIZE;
         Unsafe.getUnsafe().setMemory(recordAddress, RECORD_SIZE, (byte) 0);
         Unsafe.getUnsafe().putInt(recordAddress + MAGIC_OFFSET, FILE_MAGIC);
         Unsafe.getUnsafe().putInt(recordAddress + 4, VERSION);
@@ -286,7 +289,7 @@ private static Record readRecord(long address) {
 
     private static Record selectRecord(long address) {
         Record first = readRecord(address);
-        Record second = readRecord(address + RECORD_SIZE);
+        Record second = readRecord(address + RECORD_SLOT_SIZE);
         if (first == null) {
             return second;
         }

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

@@ -56,12 +56,15 @@
  * </ol>
  * <p>
  * On terminal failure (auth-rejection on reconnect, a cluster-wide durable-ack
- * capability gap that exhausts its settle budget, recovery error), the drainer
- * drops a {@link OrphanScanner#FAILED_SENTINEL_NAME} sentinel into the slot
- * before exiting. Future scans skip the slot until an operator clears the
- * sentinel — bounded automatic retry, then human-in-the-loop. A transient
- * all-replica failover window is NOT terminal: it is retried indefinitely
- * (Invariant B), never quarantined on a wall-clock budget or attempt cap.
+ * capability gap that exhausts its settle budget, corrupt or incomplete durable
+ * recovery state), the drainer drops a
+ * {@link OrphanScanner#FAILED_SENTINEL_NAME} sentinel into the slot before
+ * exiting. Future scans skip the slot until an operator clears the sentinel —
+ * bounded automatic retry, then human-in-the-loop. Operational setup failures
+ * leave no sentinel so a later orphan scan can retry. JVM/programming Errors
+ * also leave no sentinel and propagate after teardown. A transient all-replica
+ * failover window is NOT terminal: it is retried indefinitely (Invariant B),
+ * never quarantined on a wall-clock budget or attempt cap.
  */
 public final class BackgroundDrainer implements Runnable {
 
@@ -406,9 +409,9 @@ public WebSocketClient connectWithDurableAckRetry() {
                     // retrying cannot clear it, and spinning here would pin
                     // the slot .lock forever with no .failed sentinel and only
                     // a throttled, possibly-null-message WARN as a trace.
-                    // Rethrow: run()'s outer catch quarantines the slot
-                    // (markFailed + FAILED) and its finally releases the lock
-                    // -- quarantine-and-exit, exactly as genuine terminals do.
+                    // Rethrow: run() records the failure without attempting
+                    // allocation-heavy logging or a .failed write, its finally
+                    // releases the lock, and then the Error remains visible.
                     throw (Error) t;
                 }
                 // INVARIANT B: a transport failure -- the whole cluster is
@@ -560,15 +563,31 @@ public void run() {
                 engine = new CursorSendEngine(slotPath, segmentSizeBytes,
                         sfMaxTotalBytes, CursorSendEngine.DEFAULT_APPEND_DEADLINE_NANOS,
                         syncIntervalNanos);
-            } catch (IllegalStateException t) {
-                String msg = t.getMessage();
-                if (msg != null && msg.contains("already in use")) {
-                    LOG.info("orphan slot already locked, skipping: {} ({})",
-                            slotPath, msg);
-                    outcome = DrainOutcome.LOCKED_BY_OTHER;
-                    return;
-                }
+            } catch (SlotLockContentionException t) {
+                LOG.info("orphan slot already locked, skipping: {} ({})",
+                        slotPath, t.getMessage());
+                outcome = DrainOutcome.LOCKED_BY_OTHER;
+                return;
+            } catch (SfRecoveryException | MmapSegmentCorruptionException t) {
+                // The durable chain itself is proven corrupt or incomplete.
+                // Repeated scans cannot repair it, so preserve the terminal
+                // quarantine path in the outer catch below.
                 throw t;
+            } catch (Exception t) {
+                // Every other pre-publication construction exception is
+                // retryable: setup I/O, resource pressure, unexpected setup
+                // faults, and future operational failures do not prove durable
+                // data corruption. The constructor has closed partial
+                // resources; SlotLock retains and retries any unconfirmed
+                // unlock. Leave no .failed sentinel for the next orphan scan.
+                // Error deliberately escapes to the outer Error path: it is
+                // observable after teardown but cannot quarantine intact data.
+                String msg = t.getMessage();
+                LOG.warn("drainer setup temporarily unavailable for slot {}: {}",
+                        slotPath, msg);
+                lastErrorMessage = msg;
+                outcome = DrainOutcome.FAILED;
+                return;
             }
             long target = engine.publishedFsn();
             if (engine.ackedFsn() >= target) {
@@ -618,6 +637,16 @@ public void run() {
                     try {
                         loop.checkError();
                     } catch (Throwable t) {
+                        // The I/O loop latches a JVM/programming Error inside a
+                        // LineSenderException so checkError() can cross the
+                        // thread boundary. Preserve the original Error contract:
+                        // no wire quarantine, tear down, then propagate it.
+                        if (t instanceof Error) {
+                            throw (Error) t;
+                        }
+                        if (t.getCause() instanceof Error) {
+                            throw (Error) t.getCause();
+                        }
                         if (loop.capabilityGapTerminal() != null) {
                             // Capability gap mid-drain: recycle the wire, NOT
                             // the slot. connectWithDurableAckRetry() owns the
@@ -673,6 +702,18 @@ public void run() {
                 // outer condition, which is false for the same reason.
             }
             outcome = DrainOutcome.STOPPED;
+        } catch (Error t) {
+            // Resource pressure and JVM/programming failures prove neither
+            // durable corruption nor a terminal server response. Log the Error
+            // best-effort, but never let a secondary logging failure (especially
+            // under OOME) mask the original Error or prevent teardown.
+            try {
+                LOG.error("drainer failed with Error for slot {}", slotPath, t);
+            } catch (Throwable ignored) {
+            }
+            lastErrorMessage = t.getMessage();
+            outcome = DrainOutcome.FAILED;
+            throw t;
         } catch (Throwable t) {
             String msg = t.getMessage();
             if (slotPath != null) {

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

@@ -383,7 +383,7 @@ private CursorSendEngine(String sfDir, long segmentSizeBytes, SegmentManager man
                 // the file cannot be opened or mapped.
                 watermarkInProgress = AckWatermark.open(filesFacade, sfDir);
                 if (watermarkInProgress == null) {
-                    throw new IllegalStateException(
+                    throw new SfOperationalException(
                             "could not open required ack watermark for SF slot " + sfDir);
                 }
                 long baseSeed = lowestBase - 1;
@@ -437,7 +437,7 @@ private CursorSendEngine(String sfDir, long segmentSizeBytes, SegmentManager man
                     AckWatermark.removeOrphan(filesFacade, sfDir);
                     watermarkInProgress = AckWatermark.open(filesFacade, sfDir);
                     if (watermarkInProgress == null) {
-                        throw new IllegalStateException(
+                        throw new SfOperationalException(
                                 "could not open required ack watermark for SF slot " + sfDir);
                     }
                 }

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

@@ -30,11 +30,13 @@
  * the disk is full (the latter surfaces as backpressure on the producer
  * via {@link io.questdb.client.cutlass.line.LineSenderException}).
  * <p>
- * Recovery distinguishes two flavors: this base type marks <b>operational</b>
+ * Recovery distinguishes three flavors: this base type marks <b>operational</b>
  * failures (open/read/mmap/enumeration errors — the file's contents may be fine,
- * so recovery must fail closed), while the {@link MmapSegmentCorruptionException}
- * subtype marks <b>positively-identified corruption</b> in the file's own
- * bytes, which recovery may quarantine instead of aborting.
+ * so recovery must fail closed); {@link MmapSegmentCorruptionException} marks
+ * <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.
  */
 public class MmapSegmentException extends RuntimeException {
     public MmapSegmentException(String message) {

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

@@ -763,7 +763,7 @@ private long scanMaxGeneration(String dir) {
         if (!filesFacade.exists(dir)) return max;
         long find = filesFacade.findFirst(dir);
         if (find < 0) {
-            throw new IllegalStateException("could not enumerate SF segment directory " + dir);
+            throw new SfOperationalException("could not enumerate SF segment directory " + dir);
         }
         if (find == 0) return max;
         try {
@@ -784,7 +784,7 @@ private long scanMaxGeneration(String dir) {
                 }
             }
             if (rc < 0) {
-                throw new IllegalStateException("could not fully enumerate SF segment directory " + dir);
+                throw new SfOperationalException("could not fully enumerate SF segment directory " + dir);
             }
         } finally {
             filesFacade.findClose(find);

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

@@ -273,7 +273,7 @@ static Recovery recover(
                     // now unreadable) -- fail without mutating. Without one,
                     // legacy semantics apply: quarantine and start fresh.
                     if (manifest != null) {
-                        throw new MmapSegmentException("every SF segment in " + sfDir
+                        throw new SfRecoveryException("every SF segment in " + sfDir
                                 + " is corrupt but " + SfManifest.FILE_NAME
                                 + " references durable data");
                     }
@@ -294,7 +294,7 @@ static Recovery recover(
                     long manifestHeadBase = manifest.headBase();
                     long manifestActiveBase = manifest.activeBase();
                     if (manifestHeadBase != manifestActiveBase) {
-                        throw new MmapSegmentException(SfManifest.FILE_NAME + " in " + sfDir
+                        throw new SfRecoveryException(SfManifest.FILE_NAME + " in " + sfDir
                                 + " references durable data (headBase=" + manifestHeadBase
                                 + ", activeBase=" + manifestActiveBase
                                 + ") but no segment files exist");
@@ -328,7 +328,7 @@ static Recovery recover(
             }
             sortByBaseSeq(data, 0, data.size());
             if (manifest == null && requiresManifest) {
-                throw new MmapSegmentException("new-format SF segment exists but "
+                throw new SfRecoveryException("new-format SF segment exists but "
                         + SfManifest.FILE_NAME + " is missing");
             }
 
@@ -384,19 +384,19 @@ static Recovery recover(
                     long end = segment.baseSeq() + segment.frameCount();
                     if (segment.baseSeq() < headBase) {
                         if (end > headBase) {
-                            throw new MmapSegmentException("segment overlaps committed SF head boundary");
+                            throw new SfRecoveryException("segment overlaps committed SF head boundary");
                         }
                         continue; // acknowledged stale file after manifest-before-unlink crash
                     }
                     if (segment.baseSeq() > activeBase) {
-                        throw new MmapSegmentException("segment exists beyond committed SF active boundary");
+                        throw new SfRecoveryException("segment exists beyond committed SF active boundary");
                     }
                     chain.add(segment);
                 }
                 if (chain.size() > 0) {
                     validateContiguous(chain);
                     if (chain.get(0).baseSeq() != headBase) {
-                        throw new MmapSegmentException("missing expected SF head segment at base " + headBase);
+                        throw new SfRecoveryException("missing expected SF head segment at base " + headBase);
                     }
                 }
                 active = findActive(all, activeBase);
@@ -438,7 +438,7 @@ static Recovery recover(
                         }
                         return Recovery.empty();
                     }
-                    throw new MmapSegmentException("missing expected SF active segment at base " + activeBase);
+                    throw new SfRecoveryException("missing expected SF active segment at base " + activeBase);
                 }
                 if (chain.size() == 0) {
                     if (headBase != activeBase || active.frameCount() != 0 || corruptPaths != null) {
@@ -448,7 +448,7 @@ static Recovery recover(
                         // the same provisional baseSeq as a corrupted real
                         // active -- accepting it would quarantine unacked
                         // frames and re-issue their FSNs. Fail closed.
-                        throw new MmapSegmentException(
+                        throw new SfRecoveryException(
                                 "missing SF chain between committed boundaries"
                                         + (corruptPaths != null
                                         ? " (a corrupt segment prevents proving the empty state)" : ""));
@@ -467,7 +467,7 @@ static Recovery recover(
                         // empty-chain acceptance above).
                         chain.add(active);
                     } else {
-                        throw new MmapSegmentException(
+                        throw new SfRecoveryException(
                                 "missing expected SF active/tail segment at base " + activeBase);
                     }
                 }
@@ -483,7 +483,7 @@ static Recovery recover(
                 // independently catches sealed segments that LOST frames.
                 for (int i = 0, n = chain.size() - 1; i < n; i++) {
                     if (chain.get(i).tornTailBytes() > 0) {
-                        throw new MmapSegmentException("corrupt torn tail in sealed SF segment " + chain.get(i).path());
+                        throw new SfRecoveryException("corrupt torn tail in sealed SF segment " + chain.get(i).path());
                     }
                 }
                 for (int i = 0, n = chain.size(); i < n; i++) {
@@ -738,7 +738,7 @@ private static void validateContiguous(ObjList<MmapSegment> segments) {
             MmapSegment current = segments.get(i);
             long expected = previous.baseSeq() + previous.frameCount();
             if (current.baseSeq() != expected) {
-                throw new MmapSegmentException("FSN gap in recovered segments: expected "
+                throw new SfRecoveryException("FSN gap in recovered segments: expected "
                         + expected + " but got " + current.baseSeq());
             }
         }