fix(qwp): preserve and retire schema-rejected QWP batches

README.md

@@ -141,8 +141,63 @@ try (Sender sender = db.borrowSender()) {
 You can also let the client flush batches for you with the `auto_flush_rows` / `auto_flush_interval` config keys, e.g.
 `ws::addr=localhost:9000;auto_flush_rows=10000;auto_flush_interval=1000;`.
 
-**Confirm a batch is durably received.** Over QWP each flush returns a frame sequence number (FSN); `awaitAckedFsn`
-blocks until the server has acknowledged it. With `sf_dir`, rows in the store-and-forward log replay after reconnect
+**Schema errors and preserved copies.** With QuestDB 10.0.0 or later, QWP schema mismatches use
+`SenderError.Policy.REJECT_AND_CONTINUE`: the owning borrowed sender fails, but
+returning it and borrowing again lets the slot continue. Close and rebuild a
+standalone sender after its error. Preservation runs on the I/O thread. Slow
+disk I/O can delay close; if its
+shutdown budget expires, cleanup retains the slot lock until that thread exits,
+so rebuilding immediately may require a retry.
+Other error categories keep their existing policies. Select `.schemaMismatchPolicy(SenderError.Policy.TERMINAL)` on either
+builder to retain the old preserve-and-halt behavior.
+
+With `sf_dir`, rejected frames are copied to `<sf_dir>/<slot>/rejected/` before
+retirement. A split flush can also retire valid deferred frames preceding the
+bad frame; these are included in the copy. After restart, transaction mode is
+unknown, so retirement conservatively includes the whole recovered commit group.
+The asynchronous `errorHandler`
+receives `error.getRejectedPath()` only after the directory is complete. The
+producer exception carries the trigger FSN and affected range, but its path may
+be null while copying is pending. Notifications are retained in a separate
+256-entry queue per slot; a full queue pauses that slot's retirement. Crashes
+and shutdown can still lose queued notifications.
+
+Use `.dlqDirectory(path)` for a different base directory or a memory-only
+sender; copies go under `path/<slot>/rejected/`. Memory-only senders without an
+explicit destination retire without a preserved copy. `.dlqEnabled(false)` disables
+preservation and accepts permanent loss of retired rows. These are builder
+options. A configured destination is checked at build time; later storage
+failures pause retirement and retry the copy while keeping the source frames.
+A second schema rejection while an earlier range is pending, or an invalid
+retirement range/dictionary, logs an error and falls back to `TERMINAL`.
+Preserved payloads use the binary store-and-forward format; `rejection.properties`
+contains human-readable error metadata. A source queue namespace, the source
+segment's persisted generation token, and the exact FSN range determine the archive
+directory. A retry or restart for that same live range removes its exact crashed
+staging directory and reuses a structurally valid completed copy.
+
+Startup does not scan archive directories. If recovery finds an orphan tail, it
+checks only that range's deterministic archive path. A completed copy whose
+metadata, segment, manifest, watermark, and optional dictionary validate produces
+an asynchronous `SenderError` before the tail retires. A missing or damaged copy
+is ignored so archive output cannot block live-queue recovery. Unrelated and
+legacy `.tmp-*` directories are left untouched. A crash before publication or
+after retirement but before callback delivery can still lose the notification.
+Copy an archive to a separate working directory before replaying it, because
+normal queue cleanup removes drained data. Replay after fixing the schema can
+duplicate rows that the server committed before the error.
+
+Completed copies are never automatically deleted and can contain a full symbol dictionary
+each. Quarantining a damaged live slot also moves its archives; use the `DATA_LOSS`
+event's quarantine path to locate copies whose reported paths have moved.
+Monitor `getDlqBytesWritten()`, `getDlqFilesWritten()` and free disk space
+(the counters are available on `QwpWebSocketSender`). TLS does not encrypt these
+files at rest. With preservation disabled, a persistent schema problem can
+retire data indefinitely; keep your source data and monitor the error handler.
+
+**Wait for queue progress.** Over QWP each flush returns a frame sequence number (FSN); `awaitAckedFsn`
+blocks until that sequence is resolved. Resolution includes server acknowledgements and locally retired rejected
+frames, so observe the error handler as well; a successful wait alone does not prove every row was ingested. With `sf_dir`, rows in the store-and-forward log replay after reconnect
 or a producer-process restart. For periodic host-power-loss checkpoints, also configure
 `sf_durability=periodic;sf_sync_interval_millis=5000;`.
 
@@ -152,8 +207,8 @@ try (Sender sender = db.borrowSender()) {
         sender.table("trades").symbol("symbol", t.symbol).doubleColumn("price", t.price).atNow();
     }
     long fsn = sender.flushAndGetSequence();          // publish the batch, get its sequence number
-    if (sender.awaitAckedFsn(fsn, 30_000)) {          // block up to 30s for the server ack
-        // batch acknowledged by the server
+    if (sender.awaitAckedFsn(fsn, 30_000)) {          // block up to 30s for resolved progress
+        // queue resolved through fsn; check rejection notifications for ingestion errors
     } else {
         // not yet acked within the timeout; it stays buffered and replays on reconnect
     }

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

@@ -65,6 +65,7 @@ private static String buildMessage(SenderError e) {
         if (status != SenderError.NO_STATUS_BYTE) {
             sb.append(" (status=0x").append(Integer.toHexString(status & 0xFF)).append(')');
         }
+        sb.append(" rejectedFsn=").append(e.getRejectedFsn());
         sb.append(" fsn=[").append(e.getFromFsn()).append(',').append(e.getToFsn()).append(']');
         if (e.getTableName() != null) {
             sb.append(" table=").append(e.getTableName());

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

@@ -72,6 +72,9 @@ public final class QuestDBBuilder {
     private SenderConnectionListener connectionListener;
     private BackgroundDrainerListener drainerListener;
     private SenderErrorHandler errorHandler;
+    private SenderError.Policy schemaMismatchPolicy = SenderError.Policy.REJECT_AND_CONTINUE;
+    private boolean dlqEnabled = true;
+    private String dlqDir;
     private long housekeeperIntervalMillis = UNSET;
     private HttpTokenProvider httpTokenProvider;
     private String config;
@@ -166,6 +169,41 @@ public QuestDBBuilder errorHandler(SenderErrorHandler handler) {
         return this;
     }
 
+    /**
+         * Select schema-mismatch handling. REJECT_AND_CONTINUE fails the owning
+         * handle and retires its rejected prefix; the underlying slot continues.
+         * TERMINAL retains queued frames and halts the slot.
+         */
+    public QuestDBBuilder schemaMismatchPolicy(SenderError.Policy policy) {
+        if (policy != SenderError.Policy.TERMINAL && policy != SenderError.Policy.REJECT_AND_CONTINUE) {
+            throw new IllegalArgumentException("schema mismatch policy must be TERMINAL or REJECT_AND_CONTINUE");
+        }
+        schemaMismatchPolicy = policy;
+        return this;
+    }
+
+    /**
+         * Enable preserved copies before schema retirement (default: enabled for disk queues).
+         * Disabling preservation accepts permanent loss of retired rows.
+         */
+    public QuestDBBuilder dlqEnabled(boolean enabled) {
+        dlqEnabled = enabled;
+        return this;
+    }
+
+    /**
+         * Set the raw-copy base directory, including for memory-only queues.
+         * Copies live under directory/slot/rejected and are never automatically deleted.
+         * The asynchronous error names the completed directory.
+         */
+    public QuestDBBuilder dlqDirectory(String directory) {
+        if (directory == null || directory.isEmpty()) {
+            throw new IllegalArgumentException("DLQ directory must not be empty");
+        }
+        dlqDir = directory;
+        return this;
+    }
+
     /**
      * Builds the {@link QuestDB} handle. Validates both connect strings up
      * front -- so a malformed config fails here even when both pools have
@@ -235,10 +273,10 @@ public QuestDB build() {
                 maxLifetimeMillis,
                 housekeeperIntervalMillis,
                 queryCloseTimeoutMillis,
-                httpTokenProvider,
+                null, null, httpTokenProvider,
                 errorHandler,
                 connectionListener,
-                drainerListener
+                drainerListener, schemaMismatchPolicy, dlqEnabled, dlqDir
         );
     }
 

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

@@ -275,10 +275,16 @@ static Sender fromEnv() {
     void atNow();
 
     /**
-     * Block until the server has acknowledged every frame up to {@code targetFsn},
+     * Block until every frame up to {@code targetFsn} is resolved,
      * or until {@code timeoutMillis} elapses. Pair with {@link #flushAndGetSequence()}
      * to obtain {@code targetFsn} for a specific flush.
      * <br>
+     * Resolution includes server acknowledgements, schema-rejected ranges retired by
+     * {@link SenderError.Policy#REJECT_AND_CONTINUE}, and recovered orphan tails.
+     * A successful wait is progress, not proof that every row was ingested. A pooled
+     * sender may wait for an earlier borrow's FSN; an error owned by the current
+     * borrow still throws. Observe the error handler for earlier rejected ranges.
+     * <br>
      * When {@code request_durable_ack=on} (Enterprise primary replication), {@code targetFsn}
      * advances after durable upload to object storage, not on the ordinary commit ACK.
      * <br>
@@ -288,7 +294,7 @@ static Sender fromEnv() {
      *
      * @param targetFsn     FSN to wait for; typically the return value of {@link #flushAndGetSequence()}
      * @param timeoutMillis upper bound on the wait; {@code <= 0} returns the current state without blocking
-     * @return {@code true} if the server has acknowledged up to {@code targetFsn} on return, {@code false} on timeout
+     * @return {@code true} if the queue has resolved up to {@code targetFsn}, {@code false} on timeout
      * @throws LineSenderException if the transport has latched a terminal error
      */
     default boolean awaitAckedFsn(long targetFsn, long timeoutMillis) {
@@ -503,8 +509,8 @@ default Sender decimalColumn(CharSequence name, CharSequence value) {
      * @param timeoutMillis upper bound on the wait; {@code <= 0} returns the
      *                      current state without blocking (the flush still
      *                      happens before the check)
-     * @return {@code true} if the server has acknowledged every published
-     *         frame on return, {@code false} on timeout
+     * @return {@code true} if every published frame is resolved on return,
+     *         {@code false} on timeout
      * @throws LineSenderException if the transport has latched a terminal error
      */
     default boolean drain(long timeoutMillis) {
@@ -610,14 +616,16 @@ default Sender geoHashColumn(CharSequence name, CharSequence value) {
     }
 
     /**
-     * Highest frame sequence number (FSN) the server has acknowledged.
+     * Highest contiguous resolved frame sequence number (FSN). Includes server
+     * acknowledgements and locally retired schema-rejected ranges or orphan tails;
+     * this is queue progress, not a count of successfully ingested rows.
      * Returns {@code -1} when no batch has been published yet, and on transports that
      * do not track FSNs (HTTP, TCP, UDP).
      * <br>
      * Snapshot accessor: for a bounded blocking wait, use
      * {@link #awaitAckedFsn(long, long)}.
      *
-     * @return highest acknowledged FSN, or {@code -1} if none or unsupported
+     * @return highest resolved FSN, or {@code -1} if none or unsupported
      */
     default long getAckedFsn() {
         return -1L;
@@ -1082,6 +1090,9 @@ final class LineSenderBuilder {
         // Optional user-supplied async error handler. When null, the sender
         // uses DefaultSenderErrorHandler.INSTANCE (loud-not-silent log).
         private io.questdb.client.SenderErrorHandler errorHandler;
+        private SenderError.Policy schemaMismatchPolicy = SenderError.Policy.REJECT_AND_CONTINUE;
+        private boolean dlqEnabled = true;
+        private String dlqDir;
         // Bounded inbox capacity for the async error dispatcher.
         // PARAMETER_NOT_SET_EXPLICITLY → spec default (256).
         private int errorInboxCapacity = PARAMETER_NOT_SET_EXPLICITLY;
@@ -1714,7 +1725,8 @@ public Sender build() {
                                     actualConnectionListenerInboxCapacity,
                                     actualMaxFrameRejections,
                                     actualPoisonMinEscalationWindowMillis,
-                                    actualCatchUpCapGapMinEscalationWindowMillis
+                                    actualCatchUpCapGapMinEscalationWindowMillis,
+                                    schemaMismatchPolicy, dlqEnabled, dlqDir, transactional
                             );
                         } catch (UnreplayableSlotException e) {
                             // The one failure build() recovers from. The slot's frames reference ids
@@ -2129,11 +2141,47 @@ public LineSenderBuilder enableTls() {
             return this;
         }
 
+        /**
+         * Select schema-mismatch handling. REJECT_AND_CONTINUE fails the owning
+         * handle and retires its rejected prefix; the underlying slot continues.
+         * TERMINAL retains queued frames and halts the slot.
+         */
+        public LineSenderBuilder schemaMismatchPolicy(SenderError.Policy policy) {
+            if (policy != SenderError.Policy.TERMINAL && policy != SenderError.Policy.REJECT_AND_CONTINUE) {
+                throw new IllegalArgumentException("schema mismatch policy must be TERMINAL or REJECT_AND_CONTINUE");
+            }
+            schemaMismatchPolicy = policy;
+            return this;
+        }
+
+        /**
+         * Enable preserved copies before schema retirement (default: enabled for disk queues).
+         * Disabling preservation accepts permanent loss of retired rows.
+         */
+        public LineSenderBuilder dlqEnabled(boolean enabled) {
+            dlqEnabled = enabled;
+            return this;
+        }
+
+        /**
+         * Set the raw-copy base directory, including for memory-only queues.
+         * Copies live under directory/slot/rejected and are never automatically deleted.
+         * The asynchronous error names the completed directory.
+         */
+        public LineSenderBuilder dlqDirectory(String directory) {
+            if (directory == null || directory.isEmpty()) {
+                throw new IllegalArgumentException("DLQ directory must not be empty");
+            }
+            dlqDir = directory;
+            return this;
+        }
+
         /**
          * Sets the async error handler invoked for every server-side rejection.
          * The handler runs on a dedicated daemon dispatcher thread, never on the
          * I/O thread or producer thread. Slow handlers do not stall publishing;
-         * if the bounded inbox fills up, surplus notifications are dropped
+         * schema rejections use a separate 256-entry queue that pauses retirement when full.
+         * For other categories, if the bounded inbox fills up, surplus notifications are dropped
          * (visible via {@code QwpWebSocketSender.getDroppedErrorNotifications()}).
          *
          * <p>WebSocket transport only; setting on other transports throws.