feat(qwp): QWiP durable ack

Author: glasstiger

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

@@ -107,6 +107,8 @@ public abstract class WebSocketClient implements QuietCloseable {
     // QWP version negotiation
     private String qwpClientId;
     private int qwpMaxVersion = 1;
+    // Opt-in for STATUS_DURABLE_ACK frames; sent as X-QWP-Request-Durable-Ack: true
+    private boolean qwpRequestDurableAck;
     // Receive buffer (native memory)
     private long recvBufPtr;
     private int recvBufSize;
@@ -390,6 +392,16 @@ public void setQwpMaxVersion(int maxVersion) {
         this.qwpMaxVersion = maxVersion;
     }
 
+    /**
+     * Enables the opt-in X-QWP-Request-Durable-Ack upgrade header. When set,
+     * servers with primary replication configured will additionally emit
+     * STATUS_DURABLE_ACK frames as the WAL containing committed client
+     * messages reaches the object store.
+     */
+    public void setQwpRequestDurableAck(boolean enabled) {
+        this.qwpRequestDurableAck = enabled;
+    }
+
     /**
      * Non-blocking attempt to receive a WebSocket frame.
      * Returns immediately if no complete frame is available.
@@ -476,6 +488,9 @@ public void upgrade(CharSequence path, int timeout, CharSequence authorizationHe
             sendBuffer.putAscii(qwpClientId);
             sendBuffer.putAscii("\r\n");
         }
+        if (qwpRequestDurableAck) {
+            sendBuffer.putAscii("X-QWP-Request-Durable-Ack: true\r\n");
+        }
         if (authorizationHeader != null) {
             sendBuffer.putAscii("Authorization: ");
             sendBuffer.putAscii(authorizationHeader);

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

@@ -377,6 +377,14 @@ public Throwable getLastError() {
         return lastError.get();
     }
 
+    /**
+     * Returns the highest batch sequence acknowledged by the server, or -1 if
+     * no acknowledgment has been received yet.
+     */
+    public long getHighestAckedSequence() {
+        return highestAcked;
+    }
+
     /**
      * Returns the maximum window size.
      */

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

@@ -138,7 +138,7 @@ public class QwpWebSocketSender implements Sender {
     // Flow control
     private InFlightWindow inFlightWindow;
     private int maxSentSchemaId = -1;
-    // Track highest symbol ID sent to server (for delta encoding)
+    // Track the highest symbol ID sent to server (for delta encoding)
     // Once sent over TCP, server is guaranteed to receive it (or connection dies)
     private int maxSentSymbolId = -1;
     // Batch sequence counter (must match server's messageSequence)
@@ -147,7 +147,14 @@ public class QwpWebSocketSender implements Sender {
     // Async mode: pending row tracking
     private long pendingBytes;
     private int pendingRowCount;
+    // Highest client sequence durably uploaded, tracked when durable-ack opt-in is enabled
+    // and the server emits STATUS_DURABLE_ACK frames.
+    private long highestDurableSequence = -1;
+    // Opt-in: request server-side STATUS_DURABLE_ACK frames after WAL reaches object store.
+    // Must be set before the first send; has no effect once the WebSocket upgrade has completed.
+    private boolean requestDurableAck;
     private boolean sawBinaryAck;
+    private boolean sawPong;
     private WebSocketSendQueue sendQueue;
 
     private QwpWebSocketSender(
@@ -290,6 +297,33 @@ public static QwpWebSocketSender connect(
         return sender;
     }
 
+    public static QwpWebSocketSender connect(
+            String host,
+            int port,
+            ClientTlsConfiguration tlsConfig,
+            int autoFlushRows,
+            int autoFlushBytes,
+            long autoFlushIntervalNanos,
+            int inFlightWindowSize,
+            String authorizationHeader,
+            int maxSchemasPerConnection,
+            boolean requestDurableAck
+    ) {
+        QwpWebSocketSender sender = new QwpWebSocketSender(
+                host, port, tlsConfig,
+                autoFlushRows, autoFlushBytes, autoFlushIntervalNanos,
+                inFlightWindowSize, authorizationHeader, maxSchemasPerConnection
+        );
+        try {
+            sender.setRequestDurableAck(requestDurableAck);
+            sender.ensureConnected();
+        } catch (Throwable t) {
+            sender.close();
+            throw t;
+        }
+        return sender;
+    }
+
     /**
      * Creates a sender without connecting. For testing only.
      * <p>
@@ -302,10 +336,14 @@ public static QwpWebSocketSender connect(
      * @return unconnected sender
      */
     public static QwpWebSocketSender createForTesting(String host, int port, int inFlightWindowSize) {
+        return createForTesting(host, port, inFlightWindowSize, null);
+    }
+
+    public static QwpWebSocketSender createForTesting(String host, int port, int inFlightWindowSize, String authorizationHeader) {
         return new QwpWebSocketSender(
                 host, port, null,
                 DEFAULT_AUTO_FLUSH_ROWS, DEFAULT_AUTO_FLUSH_BYTES, DEFAULT_AUTO_FLUSH_INTERVAL_NANOS,
-                inFlightWindowSize, null, DEFAULT_MAX_SCHEMAS_PER_CONNECTION
+                inFlightWindowSize, authorizationHeader, DEFAULT_MAX_SCHEMAS_PER_CONNECTION
         );
     }
 
@@ -779,6 +817,27 @@ public int getAutoFlushRows() {
         return autoFlushRows;
     }
 
+    /**
+     * Returns the highest client batch sequence acknowledged by the server
+     * (committed to WAL), or -1 if no ACK has been received yet.
+     */
+    public long getHighestAckedSequence() {
+        return inFlightWindow != null ? inFlightWindow.getHighestAckedSequence() : -1;
+    }
+
+    /**
+     * Returns the highest client batch sequence that the server has reported
+     * as durably persisted to the object store via a STATUS_DURABLE_ACK frame,
+     * or -1 if no durable ACK has been observed yet on this connection.
+     * <p>
+     * Only meaningful when the connection was opened with
+     * {@link #setRequestDurableAck(boolean)} = true on a server where primary
+     * replication is enabled; otherwise remains -1.
+     */
+    public long getHighestDurableSequence() {
+        return sendQueue != null ? sendQueue.getHighestDurableSequence() : highestDurableSequence;
+    }
+
     /**
      * Returns the max symbol ID sent to the server.
      * Once sent over TCP, server is guaranteed to receive it (or connection dies).
@@ -960,6 +1019,27 @@ public QwpWebSocketSender longColumn(CharSequence columnName, long value) {
         return this;
     }
 
+    /**
+     * Sends a WebSocket PING and reads frames until the PONG comes back,
+     * processing any STATUS_DURABLE_ACK or STATUS_OK frames along the way.
+     * After this method returns, {@link #getHighestDurableSequence()} reflects
+     * the latest durable watermark reported by the server.
+     * <p>
+     * In async mode the PING is queued for the I/O thread; this method
+     * returns once the I/O thread has sent it and processed the response.
+     *
+     * @throws LineSenderException if the connection is closed or the ping times out
+     */
+    public void ping() {
+        checkNotClosed();
+        ensureConnected();
+        if (inFlightWindowSize > 1) {
+            sendQueue.pingAndDrain();
+        } else {
+            syncPing();
+        }
+    }
+
     @Override
     public void reset() {
         checkNotClosed();
@@ -988,6 +1068,27 @@ public void setGorillaEnabled(boolean enabled) {
         this.encoder.setGorillaEnabled(enabled);
     }
 
+    /**
+     * Opts the connection in for STATUS_DURABLE_ACK frames. Must be called
+     * before any send operation — the flag is consulted once, during WebSocket
+     * upgrade. Setting this true on a server without primary replication
+     * enabled is a no-op: the server silently ignores the header.
+     * <p>
+     * Observe durable progress via {@link #getHighestDurableSequence()}.
+     *
+     * @throws LineSenderException if the connection is already established or closed
+     */
+    public void setRequestDurableAck(boolean enabled) {
+        if (closed) {
+            throw new LineSenderException("Sender is closed");
+        }
+        if (connected) {
+            throw new LineSenderException(
+                    "setRequestDurableAck must be called before the first send");
+        }
+        this.requestDurableAck = enabled;
+    }
+
     /**
      * Adds a SHORT column value to the current row.
      *
@@ -1215,6 +1316,7 @@ private void ensureConnected() {
             try {
                 client.setQwpMaxVersion(QwpConstants.MAX_SUPPORTED_VERSION);
                 client.setQwpClientId(QwpConstants.CLIENT_ID);
+                client.setQwpRequestDurableAck(requestDurableAck);
                 client.connect(host, port);
                 client.upgrade(WRITE_PATH, authorizationHeader);
             } catch (Exception e) {
@@ -1639,6 +1741,32 @@ private boolean shouldAutoFlush() {
         return false;
     }
 
+    private void syncPing() {
+        client.sendPing(1000);
+        long deadline = System.currentTimeMillis() + InFlightWindow.DEFAULT_TIMEOUT_MS;
+        while (System.currentTimeMillis() < deadline) {
+            sawPong = false;
+            sawBinaryAck = false;
+            boolean received = client.receiveFrame(ackHandler, 1000);
+            if (received) {
+                if (sawBinaryAck) {
+                    long sequence = ackResponse.getSequence();
+                    if (ackResponse.isDurableAck()) {
+                        if (sequence > highestDurableSequence) {
+                            highestDurableSequence = sequence;
+                        }
+                    } else if (ackResponse.isSuccess()) {
+                        inFlightWindow.acknowledgeUpTo(sequence);
+                    }
+                }
+                if (sawPong) {
+                    return;
+                }
+            }
+        }
+        throw new LineSenderException("Ping timed out");
+    }
+
     private long toMicros(long value, ChronoUnit unit) {
         switch (unit) {
             case NANOS:
@@ -1696,6 +1824,13 @@ private void waitForAck(long expectedSequence) {
                             return; // Our batch was acknowledged (cumulative)
                         }
                         // Got ACK for lower sequence - continue waiting
+                    } else if (ackResponse.isDurableAck()) {
+                        // Durable-upload watermark for opted-in connections. Record the highest
+                        // value seen; this method does not block for durable acks — callers who
+                        // need to wait on durability can poll getHighestDurableSequence().
+                        if (sequence > highestDurableSequence) {
+                            highestDurableSequence = sequence;
+                        }
                     } else {
                         String errorMessage = ackResponse.getErrorMessage();
                         LineSenderException error = new LineSenderException(
@@ -1746,5 +1881,10 @@ public void onBinaryMessage(long payloadPtr, int payloadLen) {
         public void onClose(int code, String reason) {
             throw new LineSenderException("WebSocket closed while waiting for ACK: " + reason);
         }
+
+        @Override
+        public void onPong(long payloadPtr, int payloadLen) {
+            sender.sawPong = true;
+        }
     }
 }

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

@@ -58,11 +58,17 @@ public class WebSocketResponse {
     public static final int MIN_ERROR_RESPONSE_SIZE = 11; // status + sequence + error length
     // Minimum response size: status (1) + sequence (8)
     public static final int MIN_RESPONSE_SIZE = 9;
-    public static final byte STATUS_INTERNAL_ERROR = 0x06;
     // Status codes (must match QWP_SPECIFICATION.md)
     public static final byte STATUS_OK = 0x00;
-    public static final byte STATUS_PARSE_ERROR = 0x05;
+    /**
+     * Cumulative durable-upload acknowledgment. Emitted by servers where
+     * primary replication is enabled and the connection opted in via
+     * X-QWP-Request-Durable-Ack. Payload layout matches STATUS_OK.
+     */
+    public static final byte STATUS_DURABLE_ACK = 0x02;
     public static final byte STATUS_SCHEMA_MISMATCH = 0x03;
+    public static final byte STATUS_PARSE_ERROR = 0x05;
+    public static final byte STATUS_INTERNAL_ERROR = 0x06;
     public static final byte STATUS_SECURITY_ERROR = 0x08;
     public static final byte STATUS_WRITE_ERROR = 0x09;
     private String errorMessage;
@@ -75,6 +81,16 @@ public WebSocketResponse() {
         this.errorMessage = null;
     }
 
+    /**
+     * Creates a durable-upload ACK response.
+     */
+    public static WebSocketResponse durableAck(long sequence) {
+        WebSocketResponse response = new WebSocketResponse();
+        response.status = STATUS_DURABLE_ACK;
+        response.sequence = sequence;
+        return response;
+    }
+
     /**
      * Creates an error response.
      */
@@ -105,7 +121,7 @@ public static boolean isStructurallyValid(long ptr, int length) {
         }
 
         byte status = Unsafe.getUnsafe().getByte(ptr);
-        if (status == STATUS_OK) {
+        if (status == STATUS_OK || status == STATUS_DURABLE_ACK) {
             return length == MIN_RESPONSE_SIZE;
         }
 
@@ -141,13 +157,22 @@ public long getSequence() {
         return sequence;
     }
 
+    /**
+     * Returns the raw status byte.
+     */
+    public byte getStatus() {
+        return status;
+    }
+
     /**
      * Returns a human-readable status name.
      */
     public String getStatusName() {
         switch (status) {
             case STATUS_OK:
                 return "OK";
+            case STATUS_DURABLE_ACK:
+                return "DURABLE_ACK";
             case STATUS_PARSE_ERROR:
                 return "PARSE_ERROR";
             case STATUS_SCHEMA_MISMATCH:
@@ -164,7 +189,14 @@ public String getStatusName() {
     }
 
     /**
-     * Returns true if this is a success response.
+     * Returns true when this is a cumulative durable-upload ACK (STATUS_DURABLE_ACK).
+     */
+    public boolean isDurableAck() {
+        return status == STATUS_DURABLE_ACK;
+    }
+
+    /**
+     * Returns true if this is a success response (STATUS_OK).
      */
     public boolean isSuccess() {
         return status == STATUS_OK;
@@ -192,8 +224,8 @@ public boolean readFrom(long ptr, int length) {
         sequence = Unsafe.getUnsafe().getLong(ptr + offset);
         offset += 8;
 
-        // Error message (if status != OK and more data available)
-        if (status != STATUS_OK && length >= offset + 2) {
+        // Error message (present only for error frames; STATUS_OK and STATUS_DURABLE_ACK carry no message)
+        if (status != STATUS_OK && status != STATUS_DURABLE_ACK && length >= offset + 2) {
             int msgLen = Unsafe.getUnsafe().getShort(ptr + offset) & 0xFFFF;
             offset += 2;