Defer cursor SF trim to durable-ack when opted in

Add request_durable_ack=on opt-in to the cursor store-and-forward
sender. When opted in, OK frames no longer advance the trim watermark
on the on-disk SF; only STATUS_DURABLE_ACK frames do. This closes the
last data-loss window in the QWP ingestion path: with OK-driven trim,
a primary that committed to its WAL but failed before uploading to
object store could lose rows whose only remaining copy was already
trimmed from the client's SF.

The opt-in negotiates with the server through the WebSocket upgrade.
The client sends X-QWP-Request-Durable-Ack: true in the upgrade
request, and the server echoes X-QWP-Durable-Ack: enabled in the 101
response when (and only when) it has a durable-ack registry. If the
client opted in but the server did not echo the confirmation, connect
fails immediately rather than letting the SF grow until the disk
fills -- a loud failure beats silent storage exhaustion.

Trim accounting tracks per-table seqTxn watermarks in
CursorWebSocketSendLoop. OK frames enqueue (wireSeq, table list,
seqTxns) into a FIFO; durable-ack frames update watermarks and drain
the FIFO head whenever every table referenced by the head entry is at
or above its watermark. Empty OKs (zero tables) and DROP_AND_CONTINUE
rejections are trivially durable so the queue stays drained. swap
on reconnect clears the queue and watermarks; the post-reconnect
replay restores them as the server re-OKs the same wireSeq range.

Tests:
- 13 unit tests in CursorWebSocketSendLoopDurableAckTest cover OK-only
  growth, durable-ack drain, multi-table cumulative drain, partial vs
  full coverage, NACK handling, backwards-watermark rejection, and
  reconnect state reset.
- 4 integration tests in DurableAckIntegrationTest exercise the
  connect-string parser (invalid value rejected, off opts out cleanly,
  on requires server support) and end-to-end deferred-trim through a
  TestWebSocketServer.
- CursorWebSocketSendLoopDurableAckFuzzTest runs 500 iterations of
  randomised OK/durable-ack interleavings and asserts the watermark
  invariants on every iteration.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: bluestreak01

core/src/main/java/io/questdb/client/cutlass/http/client/WebSocketClient.java

@@ -75,6 +75,8 @@ public abstract class WebSocketClient implements QuietCloseable {
     private static final int PARSE_INCOMPLETE = 0;
     private static final int PARSE_NEED_MORE = -1;
     private static final int PARSE_OK = 1;
+    private static final String QWP_DURABLE_ACK_ENABLED_VALUE = "enabled";
+    private static final String QWP_DURABLE_ACK_HEADER_NAME = "X-QWP-Durable-Ack:";
     private static final String QWP_VERSION_HEADER_NAME = "X-QWP-Version:";
     private static final ThreadLocal<MessageDigest> SHA1_DIGEST = ThreadLocal.withInitial(() -> {
         try {
@@ -124,6 +126,12 @@ public abstract class WebSocketClient implements QuietCloseable {
     private int recvBufSize;
     private int recvPos;      // Write position
     private int recvReadPos;  // Read position
+    // Set during upgrade response validation when the server echoed
+    // X-QWP-Durable-Ack: enabled. Tells the sender it landed on a server that
+    // will actually emit STATUS_DURABLE_ACK frames, so its store-and-forward
+    // path can rely on durable-ack-driven trim. Absence (after opting in via
+    // setQwpRequestDurableAck) is the early-fail signal.
+    private boolean serverDurableAckEnabled;
     private int serverQwpVersion = 1;
     private boolean upgraded;
 
@@ -295,6 +303,16 @@ public boolean isConnected() {
         return upgraded && !closed && !socket.isClosed();
     }
 
+    /**
+     * Returns true when the server echoed X-QWP-Durable-Ack: enabled in the
+     * 101 upgrade response. Meaningful only after {@link #upgrade} returns;
+     * always false when the client did not opt in via
+     * {@link #setQwpRequestDurableAck}.
+     */
+    public boolean isServerDurableAckEnabled() {
+        return serverDurableAckEnabled;
+    }
+
     /**
      * Receives and processes WebSocket frames.
      *
@@ -589,6 +607,23 @@ private static boolean excludesHeaderValue(String response, String headerName, S
         return true;
     }
 
+    private static boolean extractDurableAckEnabled(String response) {
+        int headerLen = QWP_DURABLE_ACK_HEADER_NAME.length();
+        int responseLen = response.length();
+        for (int i = 0; i <= responseLen - headerLen; i++) {
+            if (response.regionMatches(true, i, QWP_DURABLE_ACK_HEADER_NAME, 0, headerLen)) {
+                int valueStart = i + headerLen;
+                int lineEnd = response.indexOf('\r', valueStart);
+                if (lineEnd < 0) {
+                    lineEnd = responseLen;
+                }
+                String value = response.substring(valueStart, lineEnd).trim();
+                return value.equalsIgnoreCase(QWP_DURABLE_ACK_ENABLED_VALUE);
+            }
+        }
+        return false;
+    }
+
     private static int extractQwpVersion(String response) {
         int headerLen = QWP_VERSION_HEADER_NAME.length();
         int responseLen = response.length();
@@ -1017,6 +1052,13 @@ private void validateUpgradeResponse(int headerEnd) {
 
         // Extract X-QWP-Version (optional, defaults to 1 if absent)
         serverQwpVersion = extractQwpVersion(response);
+
+        // Extract X-QWP-Durable-Ack confirmation (optional, absent on servers
+        // without primary replication or when the client did not opt in).
+        // Only meaningful when qwpRequestDurableAck is true; the sender
+        // checks this value to fail at connect rather than silently
+        // missing trim signals.
+        serverDurableAckEnabled = extractDurableAckEnabled(response);
     }
 
     protected void dieWaiting(int n) {

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

@@ -1766,7 +1766,8 @@ private void ensureConnected() {
                     this::buildAndConnect,
                     reconnectMaxDurationMillis,
                     reconnectInitialBackoffMillis,
-                    reconnectMaxBackoffMillis);
+                    reconnectMaxBackoffMillis,
+                    requestDurableAck);
             // Plug the async-delivery sink before start() so the I/O thread
             // never observes a null dispatcher between recordFatal and
             // notification — the test for null in dispatchError handles
@@ -1834,6 +1835,20 @@ private WebSocketClient buildAndConnect() {
             newClient.close();
             throw new LineSenderException("Failed to connect to " + host + ":" + port, e);
         }
+        // Fail at connect when the user opted into durable acks but landed on
+        // a server that did not echo the X-QWP-Durable-Ack: enabled confirmation.
+        // Without this check, store-and-forward would never receive trim signals
+        // and the on-disk store would grow unbounded -- silent storage exhaustion
+        // is a worse outcome than a loud connect-time failure.
+        if (requestDurableAck && !newClient.isServerDurableAckEnabled()) {
+            newClient.close();
+            throw new LineSenderException(
+                    "server does not support durable ack [host=" + host + ", port=" + port
+                            + "]. The client opted in via request_durable_ack=on but the server "
+                            + "did not echo X-QWP-Durable-Ack: enabled in the upgrade response. "
+                            + "Either disable request_durable_ack or connect to a server with "
+                            + "primary replication configured.");
+        }
         return newClient;
     }