feat(qwp): NACK policy v2 -- no drop, no lists, no dead senders

Replaces the two-policy NACK model (DROP_AND_CONTINUE / HALT) with three
policies that never silently lose data and never kill the sender on a
transient condition:

- RETRIABLE (WRITE_ERROR, INTERNAL_ERROR, UNKNOWN fail-open): recycle the
  connection and replay from ackedFsn+1 through the existing wire-failure
  reconnect machinery. No watermark advance, nothing dropped.
- RETRIABLE_OTHER (NOT_WRITABLE, reserved wire byte 0x0C): same, biased to
  endpoint rotation. Read-only/demotion today arrives as a role-change
  NORMAL_CLOSURE close, so the byte is reserved for future servers.
- TERMINAL (SCHEMA_MISMATCH, PARSE_ERROR, SECURITY_ERROR on a writable
  node, PROTOCOL_VIOLATION): deterministic under byte-identical replay;
  latch loudly, bytes preserved in the SF log.

WS close codes carry no policy semantics anymore: every close is
reconnect-eligible. A frame that deterministically kills the connection is
caught behaviorally by the poison-frame detector (MAX_HEAD_FRAME_REJECTIONS
consecutive rejections of the same head FSN with no ack progress escalate
to a typed PROTOCOL_VIOLATION terminal) instead of a close-code list.

Source-breaking: SenderError.Policy renames; ackedFsn now advances only on
server OKs. Design doc: design/qwp-nack-policy-v2.md.

Author: bluestreak01

.claude/skills/review-pr/SKILL.md

@@ -358,6 +358,20 @@ loses data and never hard-fails on a transient outage.
   transport error) must never increment it or burn its wall clock — a
   transient state consuming the terminal budget (shared attempt counter,
   entry-anchored deadline) IS a Critical violation of this checklist.
+- **Mid-stream server NACKs (no drop policy).** The NACK policy must mirror
+  the connect-time tiering. A rejection category that a transient cluster
+  state can produce (`WRITE_ERROR`, `INTERNAL_ERROR`, `UNKNOWN` — and any
+  future status byte) is RETRIABLE: recycle the wire and replay from
+  `ackedFsn+1`. It must NEVER drop the batch and NEVER latch a terminal /
+  quarantine a slot on first sight. Only rejections deterministic under
+  byte-identical replay (`SCHEMA_MISMATCH`, `PARSE_ERROR`, `SECURITY_ERROR`
+  on a writable node) may go TERMINAL. A client that advances the ack
+  watermark past a NACKed frame is silently losing data — Critical. A frame
+  repeatedly rejected with no ack progress must escalate through the
+  poison-frame detector (bounded consecutive strikes at the same head FSN),
+  not through a WS close-code list — close codes carry no policy semantics.
+  `UNKNOWN` must fail OPEN (retry), never closed (terminal): a status byte
+  from a newer server must degrade to retry, not to a dead sender.
 
 **Pool startup — two modes; the mode decides who sees connectivity errors.**
 - `lazy_connect=true`: `build()` MUST succeed with **no server present**. The

.pi/skills/review-pr/SKILL.md

@@ -369,6 +369,20 @@ loses data and never hard-fails on a transient outage.
   transport error) must never increment it or burn its wall clock — a
   transient state consuming the terminal budget (shared attempt counter,
   entry-anchored deadline) IS a Critical violation of this checklist.
+- **Mid-stream server NACKs (no drop policy).** The NACK policy must mirror
+  the connect-time tiering. A rejection category that a transient cluster
+  state can produce (`WRITE_ERROR`, `INTERNAL_ERROR`, `UNKNOWN` — and any
+  future status byte) is RETRIABLE: recycle the wire and replay from
+  `ackedFsn+1`. It must NEVER drop the batch and NEVER latch a terminal /
+  quarantine a slot on first sight. Only rejections deterministic under
+  byte-identical replay (`SCHEMA_MISMATCH`, `PARSE_ERROR`, `SECURITY_ERROR`
+  on a writable node) may go TERMINAL. A client that advances the ack
+  watermark past a NACKed frame is silently losing data — Critical. A frame
+  repeatedly rejected with no ack progress must escalate through the
+  poison-frame detector (bounded consecutive strikes at the same head FSN),
+  not through a WS close-code list — close codes carry no policy semantics.
+  `UNKNOWN` must fail OPEN (retry), never closed (terminal): a status byte
+  from a newer server must degrade to retry, not to a dead sender.
 
 **Pool startup — two modes; the mode decides who sees connectivity errors.**
 - `lazy_connect=true`: `build()` MUST succeed with **no server present**. The

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

@@ -30,7 +30,7 @@
 /**
  * Thrown from a producer-thread API call (typically {@link Sender#flush()}) when the
  * asynchronous SF send loop has latched a server-side rejection with policy
- * {@link SenderError.Policy#HALT}.
+ * {@link SenderError.Policy#TERMINAL}.
  *
  * <p>The wrapped {@link SenderError} carries the rejection details — category, status byte,
  * server message, FSN span, and (best-effort) table name. Use {@link #getServerError()} to

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

@@ -601,8 +601,7 @@ default Sender geoHashColumn(CharSequence name, CharSequence value) {
     }
 
     /**
-     * Highest frame sequence number (FSN) the server has acknowledged, or that the sender
-     * has skipped past on a {@link SenderError.Policy#DROP_AND_CONTINUE} rejection.
+     * Highest frame sequence number (FSN) the server has acknowledged.
      * Returns {@code -1} when no batch has been published yet, and on transports that
      * do not track FSNs (HTTP, TCP, UDP).
      * <br>

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

@@ -34,7 +34,7 @@
  * <ul>
  *   <li>Asynchronously via {@link SenderErrorHandler} registered on the builder.</li>
  *   <li>Synchronously as the payload of a {@link LineSenderServerException} thrown
- *       from the next producer-thread API call after a {@link Policy#HALT} error has
+ *       from the next producer-thread API call after a {@link Policy#TERMINAL} error has
  *       been latched.</li>
  * </ul>
  *
@@ -87,9 +87,10 @@ public SenderError(
     }
 
     /**
-     * @return the policy the I/O loop actually applied — DROP_AND_CONTINUE means the data
-     * was dropped; HALT means a {@link LineSenderServerException} will be thrown on the next
-     * producer-thread API call.
+     * @return the policy the I/O loop actually applied — RETRIABLE / RETRIABLE_OTHER means
+     * the batch stays in the store-and-forward log and is replayed after a reconnect (no data
+     * loss, informational delivery); TERMINAL means a {@link LineSenderServerException} will be
+     * thrown on the next producer-thread API call.
      */
     public @NotNull Policy getAppliedPolicy() {
         return appliedPolicy;
@@ -195,7 +196,17 @@ public enum Category {
          */
         WRITE_ERROR,
         /**
-         * WebSocket-layer close frame with a terminal code (PROTOCOL_ERROR, UNSUPPORTED_DATA, MESSAGE_TOO_BIG).
+         * Node cannot serve writes at all right now (read-only replica, primary demoting).
+         * Wire {@code 0x0C} — reserved: current servers signal this state with a
+         * reconnect-eligible close instead of a mid-stream NACK, so this category is
+         * mapped for forward compatibility with servers that NACK it explicitly.
+         */
+        NOT_WRITABLE,
+        /**
+         * A frame the server (or an intermediary) deterministically rejects: the
+         * poison-frame detector observed the same head-of-line frame fail
+         * {@link io.questdb.client.cutlass.qwp.client.sf.cursor.CursorWebSocketSendLoop#MAX_HEAD_FRAME_REJECTIONS}
+         * consecutive times with no ack progress — replaying it cannot succeed.
          */
         PROTOCOL_VIOLATION,
         /**
@@ -210,21 +221,37 @@ public enum Category {
      * connect-string per-category {@code on_*_error} → connect-string global {@code on_server_error}
      * → spec defaults.
      *
-     * <p>{@link Category#PROTOCOL_VIOLATION} and {@link Category#UNKNOWN} are forced {@link #HALT};
-     * user overrides for those categories are ignored.
+     * <p>There is no drop policy by design: the client never silently discards data. A rejected
+     * batch is either replayed ({@link #RETRIABLE} / {@link #RETRIABLE_OTHER}) or halts the
+     * sender loudly with the bytes preserved on disk ({@link #TERMINAL}).
+     *
+     * <p>{@link Category#PROTOCOL_VIOLATION} is forced {@link #TERMINAL} and
+     * {@link Category#UNKNOWN} is forced {@link #RETRIABLE} (fail open: a status byte from a
+     * newer server must degrade to retry, not to a dead sender); user overrides for those
+     * categories are ignored.
      */
     public enum Policy {
         /**
-         * Drop the rejected batch from the SF disk store (advance ackedFsn past it) and continue
-         * draining subsequent batches. The data is lost from the sender's perspective; the user
-         * must dead-letter via {@link SenderErrorHandler} if a record is needed.
+         * Recycle the connection and replay from the store-and-forward log: reconnect with
+         * capped exponential backoff and reposition at {@code ackedFsn + 1}. No data is
+         * dropped and the producer keeps writing; delivery through {@link SenderErrorHandler}
+         * is informational. A frame that keeps being rejected with no ack progress escalates
+         * to {@link #TERMINAL} via the poison-frame detector.
+         */
+        RETRIABLE,
+        /**
+         * Same replay semantics as {@link #RETRIABLE}, but the rejection says this node cannot
+         * serve writes at all (read-only replica / demoting primary), so the reconnect rotates
+         * to the next configured endpoint rather than waiting out a backoff against the same
+         * node.
          */
-        DROP_AND_CONTINUE,
+        RETRIABLE_OTHER,
         /**
          * Latch the error as terminal. The next producer-thread API call (e.g. {@link Sender#flush()})
          * throws {@link LineSenderServerException}. The sender does not drain further until the
-         * caller closes and rebuilds it.
+         * caller closes and rebuilds it. The rejected bytes remain in the store-and-forward log
+         * on disk — nothing is silently discarded.
          */
-        HALT
+        TERMINAL
     }
 }

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

@@ -44,7 +44,7 @@
  * <h2>What this callback is for</h2>
  * Dead-lettering rejected data, alerting, metrics. Producer-thread retry/abort logic
  * should not live here — that belongs in the {@code catch (LineSenderServerException)}
- * block on the producer thread, which fires after a {@link SenderError.Policy#HALT}
+ * block on the producer thread, which fires after a {@link SenderError.Policy#TERMINAL}
  * latch on the next API call.
  *
  * @see SenderError

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

@@ -30,28 +30,15 @@
  * on {@code QwpWebSocketSender} via {@code setProgressHandler(...)} or on the
  * builder via {@code LineSenderBuilder.progressHandler(...)}.
  *
- * <h2>WARNING -- settled is not durable</h2>
- * {@code ackedFsn} is a <em>settled</em> watermark, not a <em>durable</em> one.
- * A {@code DROP_AND_CONTINUE} server rejection (the default policy for the
- * {@code SCHEMA_MISMATCH} and {@code WRITE_ERROR} categories) advances
- * {@code ackedFsn} past the dropped FSN range exactly as a successful OK
- * does -- the loop cannot leave a dropped FSN unsettled without leaking
- * storage and stalling the wire. This handler therefore CANNOT distinguish
- * a batch that the server committed to the WAL from one that the server
- * discarded.
+ * <h2>Watermark advances only on server acceptance</h2>
+ * {@code ackedFsn} advances exclusively on server OK frames. A server
+ * rejection NEVER advances the watermark: retriable rejections recycle the
+ * connection and replay the batch, terminal rejections halt the sender with
+ * the bytes preserved in the store-and-forward log. There is no drop policy.
  *
- * <p>Code that gates a downstream side effect on {@code onAcked} without
- * also tracking {@link SenderErrorHandler} drops will treat dropped batches
- * as durable. The result is silent data loss: rows discarded by the server
- * are marked "saved" by the user's outbox, locks released, source records
- * deleted, downstream confirmations emitted -- for data that no longer
- * exists on the server.
- *
- * <p>Required pattern when durability matters: register a
- * {@link SenderErrorHandler}, record the {@code [fromFsn, toFsn]} range of
- * every error whose {@code getAppliedPolicy()} is
- * {@link SenderError.Policy#DROP_AND_CONTINUE}, and exclude those FSNs from
- * the "durable" set you derive from the watermark.
+ * <p>Note that in non-durable-ack mode an OK acknowledges server-side commit,
+ * not object-store durability; opt in to durable acks when replication-grade
+ * durability gates downstream side effects.
  *
  * <h2>Watermark semantics</h2>
  * The handler fires only when the watermark <em>advances</em>:
@@ -93,10 +80,9 @@
 public interface SenderProgressHandler {
     /**
      * Called when the settled watermark advances. Strictly monotonic:
-     * {@code ackedFsn} on call N+1 is greater than on call N. "Settled" covers
-     * both successful OK frames and {@code DROP_AND_CONTINUE} rejections --
-     * see the class javadoc WARNING before treating this as a durability
-     * signal.
+     * {@code ackedFsn} on call N+1 is greater than on call N. The watermark
+     * advances only on server OK frames -- rejections never advance it (see
+     * the class javadoc for durable-ack considerations).
      */
     void onAcked(long ackedFsn);
 }

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

@@ -1036,9 +1036,9 @@ public void close() {
                     //    user-installed custom handler
                     //    (hasDeliveredTerminalToCustomHandler, checked here).
                     //    The test is terminal-specific on purpose: an earlier
-                    //    routine DROP_AND_CONTINUE rejection delivered to the
-                    //    handler must NOT suppress a later genuine HALT
-                    //    terminal (the "any error ever" flag did, silently
+                    //    routine RETRIABLE rejection delivered to the
+                    //    handler must NOT suppress a later genuine TERMINAL
+                    //    error (the "any error ever" flag did, silently
                     //    losing it). It also stays false when the terminal
                     //    reached only the default handler after a
                     //    setErrorHandler(null) revert, or is still
@@ -1585,8 +1585,8 @@ public QwpWebSocketSender geoHashColumn(CharSequence columnName, CharSequence va
     }
 
     /**
-     * Highest FSN that has been server-acknowledged (or skipped past on a
-     * {@link SenderError.Policy#DROP_AND_CONTINUE} rejection). {@code -1} if
+     * Highest FSN that has been server-acknowledged. Rejections never advance
+     * the watermark. {@code -1} if
      * the I/O loop has not yet started or no batch has been published.
      * <p>
      * Snapshot accessor — for a bounded wait, use
@@ -1816,7 +1816,7 @@ public long getTotalReconnectsSucceeded() {
     }
 
     /**
-     * Total errors observed by the I/O loop (DROP and HALT combined).
+     * Total errors observed by the I/O loop (retriable and terminal combined).
      */
     public long getTotalServerErrors() {
         CursorWebSocketSendLoop l = cursorSendLoop;
@@ -2649,7 +2649,7 @@ private WebSocketClient connectWalk(ReconnectSupplier ctx) {
         // fabricate lifecycle transitions the foreground never had, steal the
         // once-per-lifetime CONNECTED classification, and re-size the
         // producer's batch guard for a connection the producer is not on
-        // (oversize batch -> ws-close[1009] -> producer-terminal HALT caused
+        // (oversize batch -> ws-close[1009] -> poison-frame escalation caused
         // by background activity).
         final boolean background = ctx.isBackground();
         // Private full-sweep cursor for background walks: claim-at-pick over

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

@@ -75,6 +75,13 @@ public class WebSocketResponse {
      */
     public static final byte STATUS_DURABLE_ACK = 0x02;
     public static final byte STATUS_INTERNAL_ERROR = 0x06;
+    /**
+     * Node cannot serve writes (read-only replica / demoting primary). Reserved:
+     * current servers signal this with a reconnect-eligible close instead of a
+     * NACK; mapped so a future server that NACKs it mid-stream gets
+     * retriable-with-rotation treatment instead of falling into UNKNOWN.
+     */
+    public static final byte STATUS_NOT_WRITABLE = 0x0C;
     // Status codes (must match https://questdb.com/docs/connect/wire-protocols/qwp-ingress-websocket/)
     public static final byte STATUS_OK = 0x00;
     public static final byte STATUS_PARSE_ERROR = 0x05;
@@ -215,6 +222,8 @@ public String getStatusName() {
                 return "SECURITY_ERROR";
             case STATUS_INTERNAL_ERROR:
                 return "INTERNAL_ERROR";
+            case STATUS_NOT_WRITABLE:
+                return "NOT_WRITABLE";
             default:
                 return "UNKNOWN(" + (status & 0xFF) + ")";
         }