Make durable-ack keepalive PING cadence configurable

bluestreak01 · claude · bluestreak01 · commit 12fc150c229d · 2026-05-05T00:56:51.000+01:00
Expose `durable_ack_keepalive_interval_millis` on the connect string and
the LineSenderBuilder. Default stays at 200 ms (matches the previously
hardcoded constant). 0 or negative disables the keepalive entirely, for
callers who send continuous traffic that already prods the server, or
who run their own ping driver.

WebSocket transport only — TCP path raises a LineSenderException if the
knob is set, mirroring how the other QWP-only knobs guard themselves.

Co-Authored-By: Claude Opus 4.7 (1M context) &lt;noreply@anthropic.com&gt;
diff --git a/core/src/main/java/io/questdb/client/Sender.java b/core/src/main/java/io/questdb/client/Sender.java
@@ -724,6 +724,12 @@ public int getTimeout() {
         private long reconnectMaxDurationMillis = PARAMETER_NOT_SET_EXPLICITLY;
         private long reconnectInitialBackoffMillis = PARAMETER_NOT_SET_EXPLICITLY;
         private long reconnectMaxBackoffMillis = PARAMETER_NOT_SET_EXPLICITLY;
+        // Cadence at which the SF cursor I/O loop sends a keepalive PING
+        // while waiting on STATUS_DURABLE_ACK frames. Default applied at
+        // build() time. 0 or negative is a documented "disable" value, so
+        // a Long.MIN_VALUE sentinel keeps it distinguishable from "unset".
+        private static final long DURABLE_ACK_KEEPALIVE_NOT_SET = Long.MIN_VALUE;
+        private long durableAckKeepaliveIntervalMillis = DURABLE_ACK_KEEPALIVE_NOT_SET;
         // Drives the initial-connect strategy. OFF is fail-fast (default).
         // SYNC retries on the user thread up to the reconnect cap. ASYNC
         // returns immediately and lets the I/O thread retry in the
@@ -1091,6 +1097,10 @@ public Sender build() {
                         reconnectMaxBackoffMillis == PARAMETER_NOT_SET_EXPLICITLY
                                 ? CursorWebSocketSendLoop.DEFAULT_RECONNECT_MAX_BACKOFF_MILLIS
                                 : reconnectMaxBackoffMillis;
+                long actualDurableAckKeepaliveIntervalMillis =
+                        durableAckKeepaliveIntervalMillis == DURABLE_ACK_KEEPALIVE_NOT_SET
+                                ? CursorWebSocketSendLoop.DEFAULT_DURABLE_ACK_KEEPALIVE_INTERVAL_MILLIS
+                                : durableAckKeepaliveIntervalMillis;
 
                 // sfDir is the parent (group root); the actual slot lives
                 // under sfDir/senderId. This is what the engine sees — the
@@ -1144,7 +1154,8 @@ public Sender build() {
                             actualReconnectMaxBackoffMillis,
                             initialConnectMode,
                             errorHandler,
-                            actualErrorInboxCapacity
+                            actualErrorInboxCapacity,
+                            actualDurableAckKeepaliveIntervalMillis
                     );
                 } catch (Throwable t) {
                     // connect() failed before ownership of cursorEngine
@@ -1934,6 +1945,32 @@ public LineSenderBuilder closeFlushTimeoutMillis(long timeoutMillis) {
             return this;
         }
 
+        /**
+         * Cadence (in millis) at which the SF cursor I/O loop sends a
+         * keepalive PING while waiting on STATUS_DURABLE_ACK frames. The
+         * server only flushes pending durable-ack frames in response to
+         * inbound recv events, so an idle opted-in client needs to prod
+         * the server periodically. Effective only when
+         * {@link #requestDurableAck(boolean) request_durable_ack=on}.
+         * <p>
+         * Pass {@code 0} or negative to disable keepalive PINGs entirely
+         * — the caller then accepts that durable-ack frames may not
+         * arrive on idle connections (e.g. they are sending data
+         * continuously, or have their own ping driver).
+         * <p>
+         * Default
+         * {@value io.questdb.client.cutlass.qwp.client.sf.cursor.CursorWebSocketSendLoop#DEFAULT_DURABLE_ACK_KEEPALIVE_INTERVAL_MILLIS}
+         * ms. WebSocket transport only.
+         */
+        public LineSenderBuilder durableAckKeepaliveIntervalMillis(long millis) {
+            if (protocol != PARAMETER_NOT_SET_EXPLICITLY && protocol != PROTOCOL_WEBSOCKET) {
+                throw new LineSenderException(
+                        "durable_ack_keepalive_interval_millis is only supported for WebSocket transport");
+            }
+            this.durableAckKeepaliveIntervalMillis = millis;
+            return this;
+        }
+
         /**
          * Per-outage cap on the cursor I/O loop's reconnect retry budget.
          * Once a wire failure occurs, the loop retries with exponential
@@ -2629,6 +2666,13 @@ private LineSenderBuilder fromConfig(CharSequence configurationString) {
                     }
                     pos = getValue(configurationString, pos, sink, "close_flush_timeout_millis");
                     closeFlushTimeoutMillis(parseLongValue(sink, "close_flush_timeout_millis"));
+                } else if (Chars.equals("durable_ack_keepalive_interval_millis", sink)) {
+                    if (protocol != PROTOCOL_WEBSOCKET) {
+                        throw new LineSenderException(
+                                "durable_ack_keepalive_interval_millis is only supported for WebSocket transport");
+                    }
+                    pos = getValue(configurationString, pos, sink, "durable_ack_keepalive_interval_millis");
+                    durableAckKeepaliveIntervalMillis(parseLongValue(sink, "durable_ack_keepalive_interval_millis"));
                 } else if (Chars.equals("reconnect_max_duration_millis", sink)) {
                     if (protocol != PROTOCOL_WEBSOCKET) {
                         throw new LineSenderException("reconnect_max_duration_millis is only supported for WebSocket transport");
diff --git a/core/src/main/java/io/questdb/client/cutlass/qwp/client/QwpWebSocketSender.java b/core/src/main/java/io/questdb/client/cutlass/qwp/client/QwpWebSocketSender.java
@@ -204,6 +204,12 @@ public class QwpWebSocketSender implements Sender {
     private long reconnectMaxDurationMillis =
             CursorWebSocketSendLoop.DEFAULT_RECONNECT_MAX_DURATION_MILLIS;
     private boolean requestDurableAck;
+    // Keepalive PING cadence used by the I/O loop while
+    // request_durable_ack=on AND there are pending durable-ack
+    // confirmations. Default mirrors the loop's spec value; 0 or negative
+    // disables keepalive PINGs entirely.
+    private long durableAckKeepaliveIntervalMillis =
+            CursorWebSocketSendLoop.DEFAULT_DURABLE_ACK_KEEPALIVE_INTERVAL_MILLIS;
 
     private QwpWebSocketSender(
             String host,
@@ -434,6 +440,42 @@ public static QwpWebSocketSender connect(
             Sender.InitialConnectMode initialConnectMode,
             SenderErrorHandler errorHandler,
             int errorInboxCapacity
+    ) {
+        return connect(host, port, tlsConfig, autoFlushRows, autoFlushBytes,
+                autoFlushIntervalNanos, inFlightWindowSize, authorizationHeader,
+                maxSchemasPerConnection, requestDurableAck, cursorEngine,
+                closeFlushTimeoutMillis, reconnectMaxDurationMillis,
+                reconnectInitialBackoffMillis, reconnectMaxBackoffMillis,
+                initialConnectMode, errorHandler, errorInboxCapacity,
+                CursorWebSocketSendLoop.DEFAULT_DURABLE_ACK_KEEPALIVE_INTERVAL_MILLIS);
+    }
+
+    /**
+     * Master connect overload — also accepts the keepalive PING cadence
+     * the I/O loop uses while waiting on STATUS_DURABLE_ACK frames.
+     * {@code 0} or negative disables the keepalive entirely (caller takes
+     * responsibility for prodding the server).
+     */
+    public static QwpWebSocketSender connect(
+            String host,
+            int port,
+            ClientTlsConfiguration tlsConfig,
+            int autoFlushRows,
+            int autoFlushBytes,
+            long autoFlushIntervalNanos,
+            int inFlightWindowSize,
+            String authorizationHeader,
+            int maxSchemasPerConnection,
+            boolean requestDurableAck,
+            CursorSendEngine cursorEngine,
+            long closeFlushTimeoutMillis,
+            long reconnectMaxDurationMillis,
+            long reconnectInitialBackoffMillis,
+            long reconnectMaxBackoffMillis,
+            Sender.InitialConnectMode initialConnectMode,
+            SenderErrorHandler errorHandler,
+            int errorInboxCapacity,
+            long durableAckKeepaliveIntervalMillis
     ) {
         QwpWebSocketSender sender = new QwpWebSocketSender(
                 host, port, tlsConfig,
@@ -446,6 +488,7 @@ public static QwpWebSocketSender connect(
             sender.reconnectMaxDurationMillis = reconnectMaxDurationMillis;
             sender.reconnectInitialBackoffMillis = reconnectInitialBackoffMillis;
             sender.reconnectMaxBackoffMillis = reconnectMaxBackoffMillis;
+            sender.durableAckKeepaliveIntervalMillis = durableAckKeepaliveIntervalMillis;
             sender.initialConnectMode = initialConnectMode == null
                     ? Sender.InitialConnectMode.OFF
                     : initialConnectMode;
@@ -1977,7 +2020,8 @@ private void ensureConnected() {
                     reconnectMaxDurationMillis,
                     reconnectInitialBackoffMillis,
                     reconnectMaxBackoffMillis,
-                    requestDurableAck);
+                    requestDurableAck,
+                    durableAckKeepaliveIntervalMillis);
             // 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
diff --git a/core/src/main/java/io/questdb/client/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoop.java b/core/src/main/java/io/questdb/client/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoop.java
@@ -80,18 +80,16 @@ public final class CursorWebSocketSendLoop implements QuietCloseable {
     /** Throttle "reconnect attempt N failed" WARN logs to one per 5 s. */
     private static final long RECONNECT_LOG_THROTTLE_NANOS = 5_000_000_000L;
     /**
-     * In durable-ack mode, the server only flushes pending STATUS_DURABLE_ACK
-     * frames in response to inbound recv events (handleBinaryMessage,
-     * handlePing, close). Without inbound traffic from the client, an idle
-     * connection that has finished publishing data never sees the ack frame
-     * even after the WAL upload has completed server-side. The I/O loop
-     * sends a WebSocket PING at this cadence whenever pendingDurable is
-     * non-empty and the loop is otherwise idle, which prods the server to
-     * call flushPendingAck and emit any progress that has accumulated. The
-     * cost is one PING per 200ms per opted-in connection, only while the
-     * client is actually waiting on durable-acks.
+     * Default cadence for the keepalive PING the I/O loop emits while
+     * waiting on STATUS_DURABLE_ACK frames. See
+     * {@link #sendDurableAckKeepaliveIfDue()} for the rationale: the OSS
+     * server only flushes pending durable-ack frames on inbound recv
+     * events, so an opted-in idle client has to prod it. {@code 200} ms
+     * trades one PING per 200 ms per idle opted-in connection for
+     * sub-second confirmation latency once the upload completes
+     * server-side. {@code 0} or negative disables the keepalive entirely.
      */
-    private static final long DURABLE_ACK_KEEPALIVE_PING_NANOS = 200_000_000L;
+    public static final long DEFAULT_DURABLE_ACK_KEEPALIVE_INTERVAL_MILLIS = 200L;
     private static final Logger LOG = LoggerFactory.getLogger(CursorWebSocketSendLoop.class);
 
     private final AtomicLong consecutiveSendErrors = new AtomicLong();
@@ -158,13 +156,15 @@ public final class CursorWebSocketSendLoop implements QuietCloseable {
     // (default), the loop trims on OK as it always has and ignores any
     // STATUS_DURABLE_ACK frames that might still arrive (logs a warning).
     private final boolean durableAckMode;
+    // Pre-converted to nanos for the comparison in sendDurableAckKeepaliveIfDue.
+    // Zero or negative disables the keepalive entirely.
+    private final long durableAckKeepaliveIntervalNanos;
     // Counters for observability of the durable-ack path. Both are zero
     // when durableAckMode is false.
     private final AtomicLong totalDurableAcks = new AtomicLong();
     private final AtomicLong totalDurableTrimAdvances = new AtomicLong();
     // Wall clock of the last keepalive PING the I/O loop sent to prod the
     // server into flushing durable-ack frames. Zero until the first PING.
-    // See DURABLE_ACK_KEEPALIVE_PING_NANOS for the rationale.
     private long lastKeepalivePingNanos;
     private WebSocketClient client;
     // fsnAtZero: FSN that wireSeq=0 maps to on the current connection. For
@@ -239,6 +239,27 @@ public CursorWebSocketSendLoop(WebSocketClient client, CursorSendEngine engine,
                                    long reconnectInitialBackoffMillis,
                                    long reconnectMaxBackoffMillis,
                                    boolean durableAckMode) {
+        this(client, engine, fsnAtZero, parkNanos, reconnectFactory,
+                reconnectMaxDurationMillis, reconnectInitialBackoffMillis,
+                reconnectMaxBackoffMillis, durableAckMode,
+                DEFAULT_DURABLE_ACK_KEEPALIVE_INTERVAL_MILLIS);
+    }
+
+    /**
+     * Master constructor — also accepts the cadence at which the I/O loop
+     * sends keepalive PINGs while waiting on STATUS_DURABLE_ACK frames.
+     * Pass {@code 0} or negative to disable keepalive PINGs entirely (the
+     * caller takes responsibility for prodding the server, e.g. by sending
+     * data, or by accepting indefinite waits on idle connections).
+     */
+    public CursorWebSocketSendLoop(WebSocketClient client, CursorSendEngine engine,
+                                   long fsnAtZero, long parkNanos,
+                                   ReconnectFactory reconnectFactory,
+                                   long reconnectMaxDurationMillis,
+                                   long reconnectInitialBackoffMillis,
+                                   long reconnectMaxBackoffMillis,
+                                   boolean durableAckMode,
+                                   long durableAckKeepaliveIntervalMillis) {
         if (engine == null) {
             throw new IllegalArgumentException("engine must be non-null");
         }
@@ -255,6 +276,9 @@ public CursorWebSocketSendLoop(WebSocketClient client, CursorSendEngine engine,
         this.reconnectInitialBackoffMillis = reconnectInitialBackoffMillis;
         this.reconnectMaxBackoffMillis = reconnectMaxBackoffMillis;
         this.durableAckMode = durableAckMode;
+        this.durableAckKeepaliveIntervalNanos = durableAckKeepaliveIntervalMillis > 0
+                ? durableAckKeepaliveIntervalMillis * 1_000_000L
+                : 0L;
         // SYNC/OFF startup hands a live client to the constructor, so we
         // already know we reached the server at least once. ASYNC startup
         // hands null and lets the I/O thread connect — hasEverConnected
@@ -888,7 +912,11 @@ private void ioLoop() {
                 //    and we have nothing else to do. The server only flushes
                 //    durable-ack frames on inbound traffic, so an idle
                 //    client otherwise never sees the WAL-upload completion.
-                if (!didWork && running && durableAckMode && !pendingDurable.isEmpty()) {
+                //    Skipped entirely when the user has disabled the
+                //    keepalive (interval <= 0).
+                if (!didWork && running && durableAckMode
+                        && durableAckKeepaliveIntervalNanos > 0L
+                        && !pendingDurable.isEmpty()) {
                     sendDurableAckKeepaliveIfDue();
                 }
                 if (!didWork && running) {
@@ -983,7 +1011,7 @@ private boolean tryReceiveAcks() {
      */
     private void sendDurableAckKeepaliveIfDue() {
         long now = System.nanoTime();
-        if (now - lastKeepalivePingNanos < DURABLE_ACK_KEEPALIVE_PING_NANOS) {
+        if (now - lastKeepalivePingNanos < durableAckKeepaliveIntervalNanos) {
             return;
         }
         lastKeepalivePingNanos = now;
diff --git a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/LineSenderBuilderWebSocketTest.java b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/LineSenderBuilderWebSocketTest.java
@@ -289,6 +289,56 @@ public void testEnableAuth_notSupported() {
                 "not supported for WebSocket");
     }
 
+    @Test
+    public void testDurableAckKeepaliveIntervalAcceptedOnWebSocket() {
+        // Positive value, zero (disable), and negative (also disable) all
+        // build cleanly. The setter is just a setter; semantics are validated
+        // at I/O-loop construction time.
+        Assert.assertNotNull(Sender.builder(Sender.Transport.WEBSOCKET)
+                .address(LOCALHOST)
+                .durableAckKeepaliveIntervalMillis(50));
+        Assert.assertNotNull(Sender.builder(Sender.Transport.WEBSOCKET)
+                .address(LOCALHOST)
+                .durableAckKeepaliveIntervalMillis(0));
+        Assert.assertNotNull(Sender.builder(Sender.Transport.WEBSOCKET)
+                .address(LOCALHOST)
+                .durableAckKeepaliveIntervalMillis(-1));
+    }
+
+    @Test
+    public void testDurableAckKeepaliveIntervalConfigStringParses() {
+        // Smoke test: the parser accepts the knob, including the disable-by-zero
+        // value, without erroring at config-string construction. The full
+        // end-to-end behaviour (server interaction) is covered by
+        // ReplicationTest in the parent repo, which exercises the default
+        // value. fromConfig() is allowed to throw a LineSenderException at
+        // connect time -- we only assert the parser did not reject the key.
+        try {
+            Sender.fromConfig("ws::addr=127.0.0.1:1;durable_ack_keepalive_interval_millis=50;");
+        } catch (LineSenderException e) {
+            Assert.assertFalse("parser must not reject the key, was: " + e.getMessage(),
+                    e.getMessage().contains("durable_ack_keepalive_interval_millis"));
+        }
+        try {
+            Sender.fromConfig("ws::addr=127.0.0.1:1;durable_ack_keepalive_interval_millis=0;");
+        } catch (LineSenderException e) {
+            Assert.assertFalse("parser must accept zero (disable), was: " + e.getMessage(),
+                    e.getMessage().contains("durable_ack_keepalive_interval_millis"));
+        }
+    }
+
+    @Test
+    public void testDurableAckKeepaliveIntervalNotSupportedForTcp() {
+        // The knob is WebSocket-only; the TCP-protocol path must reject it
+        // loudly so a user who copy-pasted the param into the wrong transport
+        // sees the mismatch immediately rather than silently losing the
+        // intended behaviour.
+        assertThrows("durable_ack_keepalive_interval_millis is only supported for WebSocket transport",
+                () -> Sender.builder(Sender.Transport.TCP)
+                        .address(LOCALHOST)
+                        .durableAckKeepaliveIntervalMillis(100));
+    }
+
     @Test
     public void testFullAsyncConfiguration() {
         Sender.LineSenderBuilder builder = Sender.builder(Sender.Transport.WEBSOCKET)
