From 352361ef891dc4cf2be2772dbbfc54c8c6b6243c Mon Sep 17 00:00:00 2001 From: Jaromir Hamala Date: Thu, 15 Oct 2026 10:07:50 +0200 Subject: [PATCH 1/9] fix(qwp): log the real resume FSN and replay count on reconnect The reconnect log said "replaying from FSN N", where N was fsnAtZero: the next unacked FSN minus any symbol-dictionary catch-up frames. On an idle connection dropped by a load balancer it read like a replay even though nothing was re-sent, and N pointed at an already-acked frame. Log the first unacked FSN and the number of frames actually replayed. --- .../client/sf/cursor/CursorWebSocketSendLoop.java | 13 +++++++++---- 1 file changed, 9 insertions(+), 4 deletions(-) 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 index 469d2a0d..aa743230 100644 --- 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 @@ -2033,12 +2033,14 @@ private void connectWalk(Throwable initial, String phase, long paceFirstAttemptM } break; } - swapClient(newClient); + long resumeFsn = swapClient(newClient); counters.reconnects.incrementAndGet(); long elapsedMs = (System.nanoTime() - outageStartNanos) / 1_000_000L; + // swapClient cleared replayTargetFsn when nothing below it is unacked + long replayFrames = replayTargetFsn >= 0L ? replayTargetFsn - resumeFsn + 1L : 0L; LOG.info("cursor I/O loop {} succeeded after {}ms, {} attempts; " - + "replaying from FSN {}", - phase, elapsedMs, attempts, externalFsnBase + fsnAtZero); + + "resuming at FSN {}, {} unacked frames to replay", + phase, elapsedMs, attempts, externalFsnBase + resumeFsn, replayFrames); return; } // A null factory result is an unsuccessful connect state, not a cap-gap @@ -2870,8 +2872,10 @@ private void snapshotReplayTarget() { * realign {@code fsnAtZero} to the next unacked FSN, restart wire * sequencing from 0, and reposition the cursor so the next * {@link #trySendOne} call replays the first unacked frame. + * + * @return the first unacked FSN, where sending resumes on the new connection */ - private void swapClient(WebSocketClient newClient) { + private long swapClient(WebSocketClient newClient) { WebSocketClient old = this.client; this.client = newClient; this.lastReconnectError = null; @@ -2915,6 +2919,7 @@ private void swapClient(WebSocketClient newClient) { // established, so the sender stays INITIALIZING and a subsequent // endpoint-policy rejection still latches the startup terminal. this.hasEverConnected = true; + return replayStart; } /** From 37e2ff2502d94e0b3132c430688b09958a294596 Mon Sep 17 00:00:00 2001 From: Jaromir Hamala Date: Thu, 15 Oct 2026 10:48:59 +0200 Subject: [PATCH 2/9] feat(qwp): send keepalive PINGs on idle connections Load balancers, proxies, firewalls, NAT gateways and the server's own idle timeout (http.net.connection.timeout, 5 min) close quiet QWP connections, so idle senders reconnected over and over. Before this, the I/O loop only pinged in durable-ack mode while acks were pending. Add keepalive_interval_millis (default 30s, 0 disables): the I/O loop pings once nothing has gone out for that long. It shares the throttle and send path with the durable-ack keepalive. --- .../main/java/io/questdb/client/Sender.java | 35 ++++++++- .../qwp/client/QwpWebSocketSender.java | 53 ++++++++++++- .../sf/cursor/CursorWebSocketSendLoop.java | 75 +++++++++++++----- .../io/questdb/client/impl/ConfigSchema.java | 1 + .../cutlass/qwp/client/KeepaliveTest.java | 77 +++++++++++++++++++ .../LineSenderBuilderWebSocketTest.java | 10 +++ .../client/QwpQueryClientFromConfigTest.java | 1 + .../qwp/websocket/TestWebSocketServer.java | 4 + .../test/impl/WsSenderConfigHonoredTest.java | 1 + 9 files changed, 234 insertions(+), 23 deletions(-) create mode 100644 core/src/test/java/io/questdb/client/test/cutlass/qwp/client/KeepaliveTest.java diff --git a/core/src/main/java/io/questdb/client/Sender.java b/core/src/main/java/io/questdb/client/Sender.java index 7617aa64..9113f2b8 100644 --- a/core/src/main/java/io/questdb/client/Sender.java +++ b/core/src/main/java/io/questdb/client/Sender.java @@ -1132,6 +1132,7 @@ final class LineSenderBuilder { private String orphanDrainBase; private int orphanDrainSlotCount; private long durableAckKeepaliveIntervalMillis = DURABLE_ACK_KEEPALIVE_NOT_SET; + private long keepaliveIntervalMillis = CursorWebSocketSendLoop.DEFAULT_KEEPALIVE_INTERVAL_MILLIS; // Optional user-supplied async error handler. When null, the sender // uses DefaultSenderErrorHandler.INSTANCE (loud-not-silent log). private io.questdb.client.SenderErrorHandler errorHandler; @@ -1716,7 +1717,8 @@ public Sender build() { actualCatchUpCapGapMinEscalationWindowMillis, symbolDictReset, actualSymbolDictResetThreshold, - actualSymbolDictResetMaxWaitMillis + actualSymbolDictResetMaxWaitMillis, + keepaliveIntervalMillis ); } catch (UnreplayableSlotException e) { // The one failure build() recovers from. The slot's frames reference ids @@ -2599,6 +2601,27 @@ public LineSenderBuilder initialConnectRetry(boolean enabled) { return this; } + /** + * Cadence (in millis) of the keepalive PING the SF cursor I/O loop + * sends once nothing has gone out on the connection for this long. + * It stops load balancers, proxies, firewalls and NAT gateways (an + * AWS ALB closes idle connections after 60 s by default), and the + * server's own idle timeout, from closing a quiet connection, which + * would otherwise cost a reconnect each time. + *

+ * Pass {@code 0} or negative to disable it. Default + * {@value io.questdb.client.cutlass.qwp.client.sf.cursor.CursorWebSocketSendLoop#DEFAULT_KEEPALIVE_INTERVAL_MILLIS} + * ms. Background orphan drainers ({@link #drainOrphans(boolean)}) + * always use the default. WebSocket transport only. + */ + public LineSenderBuilder keepaliveIntervalMillis(long millis) { + if (protocol != PARAMETER_NOT_SET_EXPLICITLY && protocol != PROTOCOL_WEBSOCKET) { + throw new LineSenderException("keepalive_interval_millis is only supported for WebSocket transport"); + } + this.keepaliveIntervalMillis = millis; + return this; + } + /** * Cap on concurrent background drainer threads when * {@link #drainOrphans(boolean)} is on. Default {@code 4}. Each @@ -4157,6 +4180,12 @@ private LineSenderBuilder fromConfig(CharSequence configurationString) { } pos = getValue(configurationString, pos, sink, "durable_ack_keepalive_interval_millis"); durableAckKeepaliveIntervalMillis(parseLongValue(sink, "durable_ack_keepalive_interval_millis")); + } else if (Chars.equals("keepalive_interval_millis", sink)) { + if (protocol != PROTOCOL_WEBSOCKET) { + throw new LineSenderException("keepalive_interval_millis is only supported for WebSocket transport"); + } + pos = getValue(configurationString, pos, sink, "keepalive_interval_millis"); + keepaliveIntervalMillis(parseLongValue(sink, "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"); @@ -4468,6 +4497,9 @@ private LineSenderBuilder fromConfigWebSocket(CharSequence configurationString) if (view.has("durable_ack_keepalive_interval_millis")) { durableAckKeepaliveIntervalMillis(wsLong(view, v, "durable_ack_keepalive_interval_millis")); } + if (view.has("keepalive_interval_millis")) { + keepaliveIntervalMillis(wsLong(view, v, "keepalive_interval_millis")); + } if (view.has("reconnect_max_duration_millis")) { reconnectMaxDurationMillis(wsLong(view, v, "reconnect_max_duration_millis")); } @@ -4677,6 +4709,7 @@ public java.util.Map wsConfigSnapshotForTest() { m.put("sf_sync_interval_millis", sfSyncIntervalMillis); m.put("close_flush_timeout_millis", closeFlushTimeoutMillis); m.put("durable_ack_keepalive_interval_millis", durableAckKeepaliveIntervalMillis); + m.put("keepalive_interval_millis", keepaliveIntervalMillis); m.put("initial_connect_retry", initialConnectMode == null ? null : initialConnectMode.name()); m.put("reconnect_max_duration_millis", reconnectMaxDurationMillis); m.put("reconnect_initial_backoff_millis", reconnectInitialBackoffMillis); 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 index f591b565..256a8897 100644 --- 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 @@ -339,6 +339,8 @@ public class QwpWebSocketSender implements Sender { // disables keepalive PINGs entirely. private long durableAckKeepaliveIntervalMillis = CursorWebSocketSendLoop.DEFAULT_DURABLE_ACK_KEEPALIVE_INTERVAL_MILLIS; + // Idle keepalive PING cadence for the I/O loop; 0 or negative disables it. + private long keepaliveIntervalMillis = CursorWebSocketSendLoop.DEFAULT_KEEPALIVE_INTERVAL_MILLIS; // Effective per-batch soft-flush threshold in raw column-buffer bytes. // Initially equals autoFlushBytes; lowered to fit under the server's // advertised X-QWP-Max-Batch-Size at handshake so the wire payload stays @@ -1223,7 +1225,7 @@ public static QwpWebSocketSender connectWithCredentialSupplier( } /** - * Master connect entry point — also accepts the poison-frame detector + * Connect entry point that also accepts the poison-frame detector * threshold ({@code max_frame_rejections}) and the symbol-dictionary * recycle knobs ({@code symbol_dict_reset}, {@code symbol_dict_reset_threshold}, * {@code symbol_dict_reset_max_wait_millis}): consecutive server-active @@ -1273,6 +1275,53 @@ public static QwpWebSocketSender connectWithCredentialSupplier( boolean symbolDictResetEnabled, int symbolDictResetThresholdSymbols, long symbolDictResetMaxWaitMillis + ) { + return connectWithCredentialSupplier(endpoints, tlsConfig, autoFlushRows, autoFlushBytes, + autoFlushIntervalNanos, authorizationHeaderSupplier, requestDurableAck, + cursorEngine, closeFlushTimeoutMillis, reconnectMaxDurationMillis, + reconnectInitialBackoffMillis, reconnectMaxBackoffMillis, + initialConnectMode, errorHandler, errorInboxCapacity, + durableAckKeepaliveIntervalMillis, authTimeoutMs, connectTimeoutMs, + connectionListener, connectionListenerInboxCapacity, + maxFrameRejections, poisonMinEscalationWindowMillis, + catchUpCapGapMinEscalationWindowMillis, + symbolDictResetEnabled, symbolDictResetThresholdSymbols, symbolDictResetMaxWaitMillis, + CursorWebSocketSendLoop.DEFAULT_KEEPALIVE_INTERVAL_MILLIS); + } + + /** + * Master connect entry point: the overload above plus the idle keepalive + * PING cadence ({@code keepalive_interval_millis}; {@code 0} or negative + * disables it). + */ + public static QwpWebSocketSender connectWithCredentialSupplier( + List endpoints, + ClientTlsConfiguration tlsConfig, + int autoFlushRows, + int autoFlushBytes, + long autoFlushIntervalNanos, + Supplier authorizationHeaderSupplier, + boolean requestDurableAck, + CursorSendEngine cursorEngine, + long closeFlushTimeoutMillis, + long reconnectMaxDurationMillis, + long reconnectInitialBackoffMillis, + long reconnectMaxBackoffMillis, + Sender.InitialConnectMode initialConnectMode, + SenderErrorHandler errorHandler, + int errorInboxCapacity, + long durableAckKeepaliveIntervalMillis, + long authTimeoutMs, + int connectTimeoutMs, + SenderConnectionListener connectionListener, + int connectionListenerInboxCapacity, + int maxFrameRejections, + long poisonMinEscalationWindowMillis, + long catchUpCapGapMinEscalationWindowMillis, + boolean symbolDictResetEnabled, + int symbolDictResetThresholdSymbols, + long symbolDictResetMaxWaitMillis, + long keepaliveIntervalMillis ) { if (symbolDictResetThresholdSymbols <= 0 || symbolDictResetThresholdSymbols > QwpConstants.MAX_SYMBOL_DICTIONARY_SIZE / 2) { @@ -1301,6 +1350,7 @@ public static QwpWebSocketSender connectWithCredentialSupplier( sender.reconnectInitialBackoffMillis = reconnectInitialBackoffMillis; sender.reconnectMaxBackoffMillis = reconnectMaxBackoffMillis; sender.durableAckKeepaliveIntervalMillis = durableAckKeepaliveIntervalMillis; + sender.keepaliveIntervalMillis = keepaliveIntervalMillis; sender.maxFrameRejections = maxFrameRejections; sender.poisonMinEscalationWindowMillis = poisonMinEscalationWindowMillis; sender.catchUpCapGapMinEscalationWindowMillis = catchUpCapGapMinEscalationWindowMillis; @@ -4896,6 +4946,7 @@ private void ensureConnected() { errorDispatcher = new SenderErrorDispatcher(errorHandler, errorInboxCapacity); } cursorSendLoop.setErrorDispatcher(errorDispatcher); + cursorSendLoop.setKeepaliveIntervalMillis(keepaliveIntervalMillis); // Symmetric progress dispatcher: lazy-allocated mirror of the // error path. Wired before start() for the same reason -- the // I/O thread should never observe a null dispatcher between 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 index aa743230..163ed9c1 100644 --- 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 @@ -129,7 +129,7 @@ public final class CursorWebSocketSendLoop implements QuietCloseable { /** * 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 + * {@link #sendPingIfDue(long)} 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 @@ -137,6 +137,17 @@ public final class CursorWebSocketSendLoop implements QuietCloseable { * server-side. {@code 0} or negative disables the keepalive entirely. */ public static final long DEFAULT_DURABLE_ACK_KEEPALIVE_INTERVAL_MILLIS = 200L; + /** + * Default cadence for the idle keepalive PING, sent once nothing has gone + * out on the wire for this long. Without it a quiet connection is closed + * by whatever sits on the path with an idle timeout -- L7 load balancers + * and proxies (AWS ALB: 60 s), firewalls, NAT gateways -- and by the + * server's own idle reaper ({@code http.net.connection.timeout}, 5 min), + * and the loop pays a reconnect each time. TCP keepalive does not help + * through an L7 proxy, which terminates TCP. {@code 0} or negative + * disables it. + */ + public static final long DEFAULT_KEEPALIVE_INTERVAL_MILLIS = 30_000L; public static final long DEFAULT_PARK_NANOS = 50_000L; // 50us idle backoff /** * Default initial reconnect backoff (100 ms). @@ -271,9 +282,11 @@ public final class CursorWebSocketSendLoop implements QuietCloseable { // visible to the loop under test. @TestOnly public static volatile boolean forceMirrorSeedFailureForTest; - // Pre-converted to nanos for the comparison in sendDurableAckKeepaliveIfDue. + // Pre-converted to nanos for the comparison in sendPingIfDue. // Zero or negative disables the keepalive entirely. private final long durableAckKeepaliveIntervalNanos; + // Idle keepalive PING cadence in nanos; 0 or negative disables. Set before start(). + private long keepaliveIntervalNanos = TimeUnit.MILLISECONDS.toNanos(DEFAULT_KEEPALIVE_INTERVAL_MILLIS); // When true, OK frames do NOT advance engine.acknowledge -- only // STATUS_DURABLE_ACK frames do. The OK frame's wireSeq is stashed in // pendingDurable along with its per-table seqTxns, and trim only advances @@ -510,10 +523,11 @@ public final class CursorWebSocketSendLoop implements QuietCloseable { // rethrows always deliver the same instance. private volatile LineSenderException terminalError; // Wall clock of the last outbound activity on the wire -- a sent frame - // (trySendOne) or a keepalive PING (sendDurableAckKeepaliveIfDue). - // Throttles the durable-ack keepalive PING so it fires only when the + // (trySendOne) or a keepalive PING (sendPingIfDue). + // Throttles the keepalive PINGs so they fire only when the // configured interval has elapsed since the most recent outbound event. - // Zero until the first send; reset to zero on reconnect. + // Set when the I/O loop starts; durable-ack mode backdates it on + // reconnect so its keepalive fires at once. private long lastFrameOrPingNanos; // Throttles the catch-up oversize WARN. Only consulted when the server // advertises NO batch cap, where the loop has no terminal to fall back on @@ -1716,6 +1730,14 @@ public void setErrorDispatcher(SenderErrorDispatcher dispatcher) { this.errorDispatcher = dispatcher; } + /** + * Idle keepalive PING cadence, see {@link #DEFAULT_KEEPALIVE_INTERVAL_MILLIS}. + * {@code 0} or negative disables it. Set once before {@link #start()}. + */ + public void setKeepaliveIntervalMillis(long millis) { + this.keepaliveIntervalNanos = TimeUnit.MILLISECONDS.toNanos(millis); + } + /** * Test seam: resize the {@link #close()} graceful-stop window (default * {@link #DEFAULT_CLOSE_GRACEFUL_STOP_MILLIS}). {@code 0} disables it, so @@ -1896,10 +1918,11 @@ private void clearDurableAckTracking() { releasePendingEntry(pendingDurable.pollFirst()); } durableTableWatermarks.clear(); - // Reset the keepalive throttle so the new connection can prod the - // server immediately rather than waiting out the leftover interval - // from before the reconnect. - lastFrameOrPingNanos = 0L; + // Backdate the keepalive throttle by one durable interval so the new + // connection can prod the server immediately rather than waiting out + // the leftover interval from before the reconnect. Relative to now, + // not 0, so the far longer idle keepalive is not also made due. + lastFrameOrPingNanos = System.nanoTime() - durableAckKeepaliveIntervalNanos; } private static boolean isCatchUpCapGap(Throwable t) { @@ -2603,6 +2626,9 @@ private void ioLoop() { if (client == null && running) { attemptInitialConnect(); } + // The upgrade handshake was outbound traffic: start the idle + // keepalive clock here rather than pinging on the first idle pass. + lastFrameOrPingNanos = System.nanoTime(); while (running) { boolean didWork = trySendOne(); // 1. Try to send next frame(s). @@ -2620,7 +2646,11 @@ private void ioLoop() { if (!didWork && running && durableAckMode && durableAckKeepaliveIntervalNanos > 0L && !pendingDurable.isEmpty()) { - sendDurableAckKeepaliveIfDue(); + sendPingIfDue(durableAckKeepaliveIntervalNanos); + } + // 4. Idle keepalive PING, see DEFAULT_KEEPALIVE_INTERVAL_MILLIS. + if (!didWork && running && keepaliveIntervalNanos > 0L) { + sendPingIfDue(keepaliveIntervalNanos); } if (!didWork && running) { LockSupport.parkNanos(parkNanos); @@ -2824,26 +2854,29 @@ private void releaseSendingSegment() { } /** - * Send a WebSocket PING to prod the server into flushing pending - * STATUS_DURABLE_ACK frames, but only when the throttle interval has - * elapsed since the last outbound activity -- a sent frame or a prior - * keepalive PING. The server's egress code only runs flushPendingAck on - * inbound recv events; without this prod, an idle connection waiting on - * durable-ack confirmation can sit forever. The "last sent frame" half - * of the gate avoids a redundant PING shortly after a producer batch - * goes idle: the recent frame already triggered a server-side flush. + * Send a WebSocket PING, but only when {@code intervalNanos} has elapsed + * since the last outbound activity -- a sent frame or a prior PING. + * Serves two callers: the idle keepalive, and the durable-ack keepalive + * that prods the server into flushing pending STATUS_DURABLE_ACK frames + * (its egress code only runs flushPendingAck on inbound recv events, so + * an idle connection waiting on durable-ack confirmation can sit + * forever). The "last sent frame" half of the gate avoids a redundant + * PING shortly after a producer batch goes idle. *

* Best-effort: any send failure routes through the standard fail() path * so the reconnect loop can take over. Caller is responsible for the - * "do we even need to send" gate (durableAckMode + non-empty pending). + * "do we even need to send" gate. */ - private void sendDurableAckKeepaliveIfDue() { + private void sendPingIfDue(long intervalNanos) { long now = System.nanoTime(); - if (now - lastFrameOrPingNanos < durableAckKeepaliveIntervalNanos) { + if (now - lastFrameOrPingNanos < intervalNanos) { return; } lastFrameOrPingNanos = now; try { + // 1 s, not the 600 s data-frame timeout: a socket that cannot take a + // 6-byte PING after an idle interval belongs to a wedged server, and + // reconnect + replay beats blocking the I/O thread for minutes. client.sendPing(1000); } catch (Throwable t) { fail(t); diff --git a/core/src/main/java/io/questdb/client/impl/ConfigSchema.java b/core/src/main/java/io/questdb/client/impl/ConfigSchema.java index b9725608..bc2d0a6a 100644 --- a/core/src/main/java/io/questdb/client/impl/ConfigSchema.java +++ b/core/src/main/java/io/questdb/client/impl/ConfigSchema.java @@ -71,6 +71,7 @@ public final class ConfigSchema { str("durable_ack_keepalive_interval_millis", Side.INGRESS); str("error_inbox_capacity", Side.INGRESS); str("initial_connect_retry", Side.INGRESS); + str("keepalive_interval_millis", Side.INGRESS); str("max_background_drainers", Side.INGRESS); // Poison-frame detector threshold (see CursorWebSocketSendLoop): the // Sender applies it; the QwpQueryClient accepts it as a syntactic no-op diff --git a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/KeepaliveTest.java b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/KeepaliveTest.java new file mode 100644 index 00000000..41c63636 --- /dev/null +++ b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/KeepaliveTest.java @@ -0,0 +1,77 @@ +/*+***************************************************************************** + * ___ _ ____ ____ + * / _ \ _ _ ___ ___| |_| _ \| __ ) + * | | | | | | |/ _ \/ __| __| | | | _ \ + * | |_| | |_| | __/\__ \ |_| |_| | |_) | + * \__\_\\__,_|\___||___/\__|____/|____/ + * + * Copyright (c) 2014-2019 Appsicle + * Copyright (c) 2019-2026 QuestDB + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + ******************************************************************************/ + +package io.questdb.client.test.cutlass.qwp.client; + +import io.questdb.client.Sender; +import io.questdb.client.test.cutlass.qwp.websocket.TestWebSocketServer; +import org.junit.Assert; +import org.junit.Test; + +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; + +public class KeepaliveTest { + + @Test + public void testIdleConnectionSendsPings() throws Exception { + AtomicInteger pings = new AtomicInteger(); + try (TestWebSocketServer server = startServer(pings); + Sender ignored = Sender.fromConfig(config(server, 100))) { + // generous: exits as soon as 3 pings land, and Windows CI can stall for seconds + long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(30); + while (pings.get() < 3 && System.nanoTime() < deadline) { + Thread.sleep(10); + } + } + Assert.assertTrue("expected periodic pings on an idle connection, saw " + pings.get(), pings.get() >= 3); + } + + @Test + public void testZeroIntervalDisablesPings() throws Exception { + AtomicInteger pings = new AtomicInteger(); + try (TestWebSocketServer server = startServer(pings); + Sender ignored = Sender.fromConfig(config(server, 0))) { + // A broken disable (interval 0 treated as "always due") pings on every idle pass + Thread.sleep(500); + } + Assert.assertEquals(0, pings.get()); + } + + private static String config(TestWebSocketServer server, long keepaliveMillis) { + return "ws::addr=localhost:" + server.getPort() + ";keepalive_interval_millis=" + keepaliveMillis + ";"; + } + + private static TestWebSocketServer startServer(AtomicInteger pings) throws Exception { + TestWebSocketServer server = new TestWebSocketServer(new TestWebSocketServer.WebSocketServerHandler() { + @Override + public void onPing(TestWebSocketServer.ClientHandler client) { + pings.incrementAndGet(); + } + }); + server.start(); + Assert.assertTrue(server.awaitStart(5, TimeUnit.SECONDS)); + return server; + } +} 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 index b3796c25..1d83dff2 100644 --- 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 @@ -511,6 +511,16 @@ public void testDurableAckKeepaliveIntervalNotSupportedForTcp() { .durableAckKeepaliveIntervalMillis(100)); } + @Test + public void testKeepaliveIntervalNotSupportedForTcp() { + assertThrows("keepalive_interval_millis is only supported for WebSocket transport", + () -> Sender.builder(Sender.Transport.TCP) + .address(LOCALHOST) + .keepaliveIntervalMillis(100)); + assertThrows("keepalive_interval_millis is only supported for WebSocket transport", + () -> Sender.fromConfig("tcp::addr=localhost:9009;keepalive_interval_millis=100;")); + } + @Test public void testAuthTimeoutConfig_acceptsPositive() { Sender.LineSenderBuilder builder = Sender.builder("ws::addr=localhost:9000;auth_timeout_ms=2500;"); diff --git a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/QwpQueryClientFromConfigTest.java b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/QwpQueryClientFromConfigTest.java index daf55bb9..1de946c4 100644 --- a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/QwpQueryClientFromConfigTest.java +++ b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/QwpQueryClientFromConfigTest.java @@ -712,6 +712,7 @@ public void testIngressOnlyKeysSilentlyAcceptedOnEgress() { "durable_ack_keepalive_interval_millis=200", "error_inbox_capacity=256", "initial_connect_retry=on", + "keepalive_interval_millis=30000", "max_background_drainers=4", "max_name_len=127", "reconnect_initial_backoff_millis=100", diff --git a/core/src/test/java/io/questdb/client/test/cutlass/qwp/websocket/TestWebSocketServer.java b/core/src/test/java/io/questdb/client/test/cutlass/qwp/websocket/TestWebSocketServer.java index ed450c5c..41d7cb41 100644 --- a/core/src/test/java/io/questdb/client/test/cutlass/qwp/websocket/TestWebSocketServer.java +++ b/core/src/test/java/io/questdb/client/test/cutlass/qwp/websocket/TestWebSocketServer.java @@ -439,6 +439,9 @@ public void start() throws IOException { public interface WebSocketServerHandler { default void onBinaryMessage(ClientHandler client, byte[] data) { } + + default void onPing(ClientHandler client) { + } } /** @@ -601,6 +604,7 @@ private void handleRead() { handler.onBinaryMessage(this, payload); break; case WebSocketOpcode.PING: + handler.onPing(this); try { writeFrame(WebSocketOpcode.PONG, payload, payload.length); } catch (IOException e) { diff --git a/core/src/test/java/io/questdb/client/test/impl/WsSenderConfigHonoredTest.java b/core/src/test/java/io/questdb/client/test/impl/WsSenderConfigHonoredTest.java index 771179ca..c7a48724 100644 --- a/core/src/test/java/io/questdb/client/test/impl/WsSenderConfigHonoredTest.java +++ b/core/src/test/java/io/questdb/client/test/impl/WsSenderConfigHonoredTest.java @@ -69,6 +69,7 @@ public void testEveryIngressKeyIsHonored() { assertHonored("close_flush_timeout_millis=2500", "close_flush_timeout_millis", 2500L); assertHonored("durable_ack_keepalive_interval_millis=900", "durable_ack_keepalive_interval_millis", 900L); assertHonored("initial_connect_retry=async", "initial_connect_retry", "ASYNC"); + assertHonored("keepalive_interval_millis=900", "keepalive_interval_millis", 900L); assertHonored("reconnect_max_duration_millis=12345", "reconnect_max_duration_millis", 12345L); assertHonored("reconnect_initial_backoff_millis=111", "reconnect_initial_backoff_millis", 111L); assertHonored("reconnect_max_backoff_millis=2222", "reconnect_max_backoff_millis", 2222L); From c96375facb2eed1a9f7afe4a42272ae4cad627c8 Mon Sep 17 00:00:00 2001 From: Jaromir Hamala Date: Thu, 15 Oct 2026 11:14:53 +0200 Subject: [PATCH 3/9] refactor(qwp): drop keepalive factory overload and ExportedApiCompatibilityTest Only Sender and its builder are public API. The test pinned internal signatures (QwpWebSocketSender factories, CursorWebSocketSendLoop constructors) whose only callers live in this jar, so they can never link against a different version. Widen the existing factory in place instead of adding an overload. --- .../qwp/client/QwpWebSocketSender.java | 64 +---- .../sf/cursor/CursorWebSocketSendLoop.java | 2 +- .../compat/ExportedApiCompatibilityTest.java | 270 ------------------ 3 files changed, 12 insertions(+), 324 deletions(-) delete mode 100644 core/src/test/java/io/questdb/client/test/compat/ExportedApiCompatibilityTest.java 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 index 256a8897..2892ef67 100644 --- 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 @@ -1049,7 +1049,8 @@ public static QwpWebSocketSender connectWithCredentialSupplier( CursorWebSocketSendLoop.DEFAULT_CATCHUP_CAP_GAP_MIN_ESCALATION_WINDOW_MILLIS, DEFAULT_SYMBOL_DICT_RESET_ENABLED, DEFAULT_SYMBOL_DICT_RESET_THRESHOLD_SYMBOLS, - DEFAULT_SYMBOL_DICT_RESET_MAX_WAIT_MILLIS); + DEFAULT_SYMBOL_DICT_RESET_MAX_WAIT_MILLIS, + CursorWebSocketSendLoop.DEFAULT_KEEPALIVE_INTERVAL_MILLIS); } /** @@ -1107,7 +1108,8 @@ autoFlushIntervalNanos, fixedAuthHeader(authorizationHeader), catchUpCapGapMinEscalationWindowMillis, DEFAULT_SYMBOL_DICT_RESET_ENABLED, DEFAULT_SYMBOL_DICT_RESET_THRESHOLD_SYMBOLS, - DEFAULT_SYMBOL_DICT_RESET_MAX_WAIT_MILLIS); + DEFAULT_SYMBOL_DICT_RESET_MAX_WAIT_MILLIS, + CursorWebSocketSendLoop.DEFAULT_KEEPALIVE_INTERVAL_MILLIS); } /** @@ -1168,7 +1170,7 @@ autoFlushIntervalNanos, fixedAuthHeader(authorizationHeader), maxFrameRejections, poisonMinEscalationWindowMillis, catchUpCapGapMinEscalationWindowMillis, symbolDictResetEnabled, symbolDictResetThresholdSymbols, - symbolDictResetMaxWaitMillis); + symbolDictResetMaxWaitMillis, CursorWebSocketSendLoop.DEFAULT_KEEPALIVE_INTERVAL_MILLIS); } /** @@ -1178,7 +1180,7 @@ autoFlushIntervalNanos, fixedAuthHeader(authorizationHeader), * long, long, long, long, Sender.InitialConnectMode, SenderErrorHandler, int, long, long, int, * SenderConnectionListener, int, int, long, long)}. Kept as its own overload, delegating to the master * below with the symbol-dictionary recycle knobs defaulted, so a caller compiled before those knobs' - * arity moved into the master keeps linking; {@code ExportedApiCompatibilityTest} pins it. + * arity moved into the master keeps linking. *

* Recycling requires a non-null engine rebuild factory installed via * {@link #setEngineRebuildFactory(EngineRebuildFactory)}; @@ -1221,16 +1223,18 @@ public static QwpWebSocketSender connectWithCredentialSupplier( catchUpCapGapMinEscalationWindowMillis, DEFAULT_SYMBOL_DICT_RESET_ENABLED, DEFAULT_SYMBOL_DICT_RESET_THRESHOLD_SYMBOLS, - DEFAULT_SYMBOL_DICT_RESET_MAX_WAIT_MILLIS); + DEFAULT_SYMBOL_DICT_RESET_MAX_WAIT_MILLIS, + CursorWebSocketSendLoop.DEFAULT_KEEPALIVE_INTERVAL_MILLIS); } /** - * Connect entry point that also accepts the poison-frame detector + * Master connect entry point — also accepts the poison-frame detector * threshold ({@code max_frame_rejections}) and the symbol-dictionary * recycle knobs ({@code symbol_dict_reset}, {@code symbol_dict_reset_threshold}, * {@code symbol_dict_reset_max_wait_millis}): consecutive server-active * rejections of the same head-of-line frame, with no ack progress in - * between, before the loop escalates to a typed terminal. + * between, before the loop escalates to a typed terminal. Also takes the + * idle keepalive PING cadence ({@code keepalive_interval_millis}). *

* Named apart from {@code connect} for the reason given on * {@link #connectWithCredentialSupplier(List, ClientTlsConfiguration, int, int, long, Supplier, @@ -1248,52 +1252,6 @@ public static QwpWebSocketSender connectWithCredentialSupplier( * * @throws LineSenderException if either recycle setting is out of range */ - public static QwpWebSocketSender connectWithCredentialSupplier( - List endpoints, - ClientTlsConfiguration tlsConfig, - int autoFlushRows, - int autoFlushBytes, - long autoFlushIntervalNanos, - Supplier authorizationHeaderSupplier, - boolean requestDurableAck, - CursorSendEngine cursorEngine, - long closeFlushTimeoutMillis, - long reconnectMaxDurationMillis, - long reconnectInitialBackoffMillis, - long reconnectMaxBackoffMillis, - Sender.InitialConnectMode initialConnectMode, - SenderErrorHandler errorHandler, - int errorInboxCapacity, - long durableAckKeepaliveIntervalMillis, - long authTimeoutMs, - int connectTimeoutMs, - SenderConnectionListener connectionListener, - int connectionListenerInboxCapacity, - int maxFrameRejections, - long poisonMinEscalationWindowMillis, - long catchUpCapGapMinEscalationWindowMillis, - boolean symbolDictResetEnabled, - int symbolDictResetThresholdSymbols, - long symbolDictResetMaxWaitMillis - ) { - return connectWithCredentialSupplier(endpoints, tlsConfig, autoFlushRows, autoFlushBytes, - autoFlushIntervalNanos, authorizationHeaderSupplier, requestDurableAck, - cursorEngine, closeFlushTimeoutMillis, reconnectMaxDurationMillis, - reconnectInitialBackoffMillis, reconnectMaxBackoffMillis, - initialConnectMode, errorHandler, errorInboxCapacity, - durableAckKeepaliveIntervalMillis, authTimeoutMs, connectTimeoutMs, - connectionListener, connectionListenerInboxCapacity, - maxFrameRejections, poisonMinEscalationWindowMillis, - catchUpCapGapMinEscalationWindowMillis, - symbolDictResetEnabled, symbolDictResetThresholdSymbols, symbolDictResetMaxWaitMillis, - CursorWebSocketSendLoop.DEFAULT_KEEPALIVE_INTERVAL_MILLIS); - } - - /** - * Master connect entry point: the overload above plus the idle keepalive - * PING cadence ({@code keepalive_interval_millis}; {@code 0} or negative - * disables it). - */ public static QwpWebSocketSender connectWithCredentialSupplier( List endpoints, ClientTlsConfiguration tlsConfig, 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 index 163ed9c1..a5256932 100644 --- 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 @@ -983,7 +983,7 @@ private CursorWebSocketSendLoop(WebSocketClient client, CursorSendEngine engine, * The policy-aware constructor as released in 1.3.x: equivalent to the overload * below with {@code externalFsnBase = 0L}, the value every caller outside a * symbol-dictionary recycle passes. Kept so callers compiled against a released - * jar keep linking; {@code ExportedApiCompatibilityTest} pins it. + * jar keep linking. */ public CursorWebSocketSendLoop(WebSocketClient client, CursorSendEngine engine, long fsnAtZero, long parkNanos, diff --git a/core/src/test/java/io/questdb/client/test/compat/ExportedApiCompatibilityTest.java b/core/src/test/java/io/questdb/client/test/compat/ExportedApiCompatibilityTest.java deleted file mode 100644 index 48e7e864..00000000 --- a/core/src/test/java/io/questdb/client/test/compat/ExportedApiCompatibilityTest.java +++ /dev/null @@ -1,270 +0,0 @@ -/*+***************************************************************************** - * ___ _ ____ ____ - * / _ \ _ _ ___ ___| |_| _ \| __ ) - * | | | | | | |/ _ \/ __| __| | | | _ \ - * | |_| | |_| | __/\__ \ |_| |_| | |_) | - * \__\_\\__,_|\___||___/\__|____/|____/ - * - * Copyright (c) 2014-2019 Appsicle - * Copyright (c) 2019-2026 QuestDB - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - * - ******************************************************************************/ - -package io.questdb.client.test.compat; - -import io.questdb.client.cutlass.http.client.Fragment; -import io.questdb.client.cutlass.http.client.Response; -import io.questdb.client.cutlass.line.http.AbstractLineHttpSender; -import io.questdb.client.cutlass.qwp.client.QwpWebSocketSender; -import io.questdb.client.cutlass.qwp.client.sf.cursor.CursorWebSocketSendLoop; -import org.junit.Assert; -import org.junit.Test; - -import java.lang.reflect.Constructor; -import java.lang.reflect.Method; -import java.lang.reflect.Modifier; -import java.util.LinkedHashSet; -import java.util.Set; -import java.util.TreeSet; - -/** - * Pins the public signatures this branch had to restore after replacing them in place. - *

- * Four exported methods were changed rather than added to - {@code Response.recv(int)} arrived as an - * abstract interface method, two {@code QwpWebSocketSender.connect(..., String, ...)} overloads were retyped - * to {@code Supplier}, the multi-host {@code AbstractLineHttpSender.createLineSender} gained a - * parameter in place, and the twenty-three-arg {@code QwpWebSocketSender.connectWithCredentialSupplier} - * overload gained the symbol-dictionary recycle knobs in place instead of through a new overload. All four - * sit in packages {@code module-info.java} exports and that ship a javadoc jar, so a caller compiled against - * an earlier release - or, for the credential-supplier overload, against main before this branch - would - * have failed with {@code NoSuchMethodError}, and an external {@code Response} implementation with - * {@code AbstractMethodError}. Nothing in this repository, in questdb, or in questdb-enterprise calls them, - * which is why the break was latent rather than observed - and why nothing would have caught it coming back. - *

- * There is no japicmp or revapi gate on this build, so this test is the gate. The expected signatures below - * are the ones present at this branch's merge base ({@code 981bdb02}); they are written out literally rather - * than derived from the current classes, because a pin computed from the thing it pins proves nothing. - * Adding an overload is fine and this test stays green; retyping or removing one turns it red. - *

- * Not every pin guards a signature that shipped. The {@code connect}, {@code createLineSender} and - * {@code Response.recv} pins, and the seven {@link #RELEASED_CONSTRUCTOR_SIGNATURES} constructor pins, all - * guard signatures present in a published jar - 1.3.9 has every one of them. The two - * {@code connectWithCredentialSupplier} pins are different: that signature is on {@code main} since the OIDC - * device-flow change ({@code 0b9b5766}), after 1.3.9, and has not shipped in a release, so the pin protects - * a caller building against {@code main}, not against a published jar. - *

- * {@link #RELEASED_CONSTRUCTOR_SIGNATURES} pins every public {@code CursorWebSocketSendLoop} constructor - * present in the published {@code org.questdb:questdb-client:1.3.9} jar, copied from {@code javap} of that - * jar for the same reason: the policy-aware constructor was widened in place on this branch and the - * released 13-argument form had to be restored as an overload. - */ -public class ExportedApiCompatibilityTest { - - /** - * Every {@code QwpWebSocketSender.connect}, {@code QwpWebSocketSender.connectWithCredentialSupplier} and - * {@code AbstractLineHttpSender.createLineSender} signature that existed at the merge base, as - * {@code name(paramType,...)returnType} over erased type names. - */ - private static final String[] PRE_BRANCH_SIGNATURES = { - // ---- QwpWebSocketSender.connect ---- - "connect(java.lang.String,int)io.questdb.client.cutlass.qwp.client.QwpWebSocketSender", - "connect(java.lang.String,int,io.questdb.client.ClientTlsConfiguration)io.questdb.client.cutlass.qwp.client.QwpWebSocketSender", - "connect(java.lang.String,int,io.questdb.client.ClientTlsConfiguration,int,int,long,java.lang.String,boolean,io.questdb.client.cutlass.qwp.client.sf.cursor.CursorSendEngine)io.questdb.client.cutlass.qwp.client.QwpWebSocketSender", - "connect(java.lang.String,int,io.questdb.client.ClientTlsConfiguration,int,int,long,java.lang.String,boolean,io.questdb.client.cutlass.qwp.client.sf.cursor.CursorSendEngine,long)io.questdb.client.cutlass.qwp.client.QwpWebSocketSender", - "connect(java.lang.String,int,io.questdb.client.ClientTlsConfiguration,int,int,long,java.lang.String,boolean,io.questdb.client.cutlass.qwp.client.sf.cursor.CursorSendEngine,long,long,long,long)io.questdb.client.cutlass.qwp.client.QwpWebSocketSender", - "connect(java.lang.String,int,io.questdb.client.ClientTlsConfiguration,int,int,long,java.lang.String,boolean,io.questdb.client.cutlass.qwp.client.sf.cursor.CursorSendEngine,long,long,long,long,io.questdb.client.Sender$InitialConnectMode)io.questdb.client.cutlass.qwp.client.QwpWebSocketSender", - "connect(java.lang.String,int,io.questdb.client.ClientTlsConfiguration,int,int,long,java.lang.String,boolean,io.questdb.client.cutlass.qwp.client.sf.cursor.CursorSendEngine,long,long,long,long,io.questdb.client.Sender$InitialConnectMode,io.questdb.client.SenderErrorHandler,int)io.questdb.client.cutlass.qwp.client.QwpWebSocketSender", - "connect(java.lang.String,int,io.questdb.client.ClientTlsConfiguration,int,int,long,java.lang.String,boolean,io.questdb.client.cutlass.qwp.client.sf.cursor.CursorSendEngine,long,long,long,long,io.questdb.client.Sender$InitialConnectMode,io.questdb.client.SenderErrorHandler,int,long)io.questdb.client.cutlass.qwp.client.QwpWebSocketSender", - "connect(java.util.List,io.questdb.client.ClientTlsConfiguration,int,int,long,java.lang.String,boolean,io.questdb.client.cutlass.qwp.client.sf.cursor.CursorSendEngine,long,long,long,long,io.questdb.client.Sender$InitialConnectMode,io.questdb.client.SenderErrorHandler,int,long,long)io.questdb.client.cutlass.qwp.client.QwpWebSocketSender", - "connect(java.util.List,io.questdb.client.ClientTlsConfiguration,int,int,long,java.lang.String,boolean,io.questdb.client.cutlass.qwp.client.sf.cursor.CursorSendEngine,long,long,long,long,io.questdb.client.Sender$InitialConnectMode,io.questdb.client.SenderErrorHandler,int,long,long,int,io.questdb.client.SenderConnectionListener,int)io.questdb.client.cutlass.qwp.client.QwpWebSocketSender", - "connect(java.util.List,io.questdb.client.ClientTlsConfiguration,int,int,long,java.lang.String,boolean,io.questdb.client.cutlass.qwp.client.sf.cursor.CursorSendEngine,long,long,long,long,io.questdb.client.Sender$InitialConnectMode,io.questdb.client.SenderErrorHandler,int,long,long,int,io.questdb.client.SenderConnectionListener,int,int,long,long)io.questdb.client.cutlass.qwp.client.QwpWebSocketSender", - // ---- QwpWebSocketSender.connectWithCredentialSupplier ---- - "connectWithCredentialSupplier(java.util.List,io.questdb.client.ClientTlsConfiguration,int,int,long,java.util.function.Supplier,boolean,io.questdb.client.cutlass.qwp.client.sf.cursor.CursorSendEngine,long,long,long,long,io.questdb.client.Sender$InitialConnectMode,io.questdb.client.SenderErrorHandler,int,long,long,int,io.questdb.client.SenderConnectionListener,int)io.questdb.client.cutlass.qwp.client.QwpWebSocketSender", - "connectWithCredentialSupplier(java.util.List,io.questdb.client.ClientTlsConfiguration,int,int,long,java.util.function.Supplier,boolean,io.questdb.client.cutlass.qwp.client.sf.cursor.CursorSendEngine,long,long,long,long,io.questdb.client.Sender$InitialConnectMode,io.questdb.client.SenderErrorHandler,int,long,long,int,io.questdb.client.SenderConnectionListener,int,int,long,long)io.questdb.client.cutlass.qwp.client.QwpWebSocketSender", - // ---- AbstractLineHttpSender.createLineSender ---- - "createLineSender(java.lang.String,int,java.lang.String,io.questdb.client.HttpClientConfiguration,io.questdb.client.ClientTlsConfiguration,int,java.lang.String,java.lang.String,java.lang.String,int,long,int,long,long,int)io.questdb.client.cutlass.line.http.AbstractLineHttpSender", - "createLineSender(io.questdb.client.std.ObjList,io.questdb.client.std.IntList,java.lang.String,io.questdb.client.HttpClientConfiguration,io.questdb.client.ClientTlsConfiguration,int,java.lang.String,java.lang.String,java.lang.String,int,long,int,long,long,int)io.questdb.client.cutlass.line.http.AbstractLineHttpSender", - }; - - /** - * Every public {@code CursorWebSocketSendLoop} constructor in the 1.3.9 jar, as {@code (paramType,...)} - * over erased type names, in the order {@code javap} printed them. - */ - private static final String[] RELEASED_CONSTRUCTOR_SIGNATURES = { - "(io.questdb.client.cutlass.http.client.WebSocketClient,io.questdb.client.cutlass.qwp.client.sf.cursor.CursorSendEngine,long,long,io.questdb.client.cutlass.qwp.client.sf.cursor.CursorWebSocketSendLoop$ReconnectFactory,long,long)", - "(io.questdb.client.cutlass.http.client.WebSocketClient,io.questdb.client.cutlass.qwp.client.sf.cursor.CursorSendEngine,long,long,io.questdb.client.cutlass.qwp.client.sf.cursor.CursorWebSocketSendLoop$ReconnectFactory,long,long,boolean)", - "(io.questdb.client.cutlass.http.client.WebSocketClient,io.questdb.client.cutlass.qwp.client.sf.cursor.CursorSendEngine,long,long,io.questdb.client.cutlass.qwp.client.sf.cursor.CursorWebSocketSendLoop$ReconnectFactory,long,long,boolean,long)", - "(io.questdb.client.cutlass.http.client.WebSocketClient,io.questdb.client.cutlass.qwp.client.sf.cursor.CursorSendEngine,long,long,io.questdb.client.cutlass.qwp.client.sf.cursor.CursorWebSocketSendLoop$ReconnectFactory,long,long,boolean,long,int)", - "(io.questdb.client.cutlass.http.client.WebSocketClient,io.questdb.client.cutlass.qwp.client.sf.cursor.CursorSendEngine,long,long,io.questdb.client.cutlass.qwp.client.sf.cursor.CursorWebSocketSendLoop$ReconnectFactory,long,long,boolean,long,int,long)", - "(io.questdb.client.cutlass.http.client.WebSocketClient,io.questdb.client.cutlass.qwp.client.sf.cursor.CursorSendEngine,long,long,io.questdb.client.cutlass.qwp.client.sf.cursor.CursorWebSocketSendLoop$ReconnectFactory,long,long,boolean,long,int,long,long)", - "(io.questdb.client.cutlass.http.client.WebSocketClient,io.questdb.client.cutlass.qwp.client.sf.cursor.CursorSendEngine,long,long,io.questdb.client.cutlass.qwp.client.sf.cursor.CursorWebSocketSendLoop$ReconnectFactory,long,long,boolean,long,int,long,long,io.questdb.client.cutlass.qwp.client.sf.cursor.CursorWebSocketSendLoop$ReconnectPolicy)", - }; - - @Test - public void testPreBranchCreateLineSenderOverloadsStillLink() { - assertSignaturesPresent(AbstractLineHttpSender.class, "createLineSender"); - } - - @Test - public void testPreBranchQwpWebSocketSenderConnectOverloadsStillLink() { - assertSignaturesPresent(QwpWebSocketSender.class, "connect"); - } - - @Test - public void testPreBranchQwpWebSocketSenderConnectWithCredentialSupplierOverloadsStillLink() { - assertSignaturesPresent(QwpWebSocketSender.class, "connectWithCredentialSupplier"); - } - - @Test - public void testReleasedCursorWebSocketSendLoopConstructorsStillLink() { - assertConstructorsPresent(CursorWebSocketSendLoop.class, RELEASED_CONSTRUCTOR_SIGNATURES); - } - - @Test - public void testResponseRecvIntIsDefaultNotAbstract() throws Exception { - // The defect was recv(int) arriving as an abstract interface method: an external implementation - // written against the earlier Response compiles fine and then fails at run time with - // AbstractMethodError, which is exactly the failure a unit test of this library never sees. - Method recvWithTimeout = Response.class.getMethod("recv", int.class); - Assert.assertFalse("Response.recv(int) must stay a default method - an implementation written " - + "before the overload existed has no override for it", - Modifier.isAbstract(recvWithTimeout.getModifiers())); - Assert.assertEquals(Fragment.class, recvWithTimeout.getReturnType()); - - Method recv = Response.class.getMethod("recv"); - Assert.assertTrue("recv() is the one method an implementation must supply", - Modifier.isAbstract(recv.getModifiers())); - Assert.assertEquals(Fragment.class, recv.getReturnType()); - } - - @Test - public void testResponseImplementorOverridingOnlyRecvStillWorks() { - // LegacyResponse below is the compile-time half: it implements Response and overrides recv() ONLY, - // exactly as an implementation predating the overload does. If recv(int) went back to being - // abstract, this class stops compiling and the whole test module goes with it - which is the point. - LegacyResponse legacy = new LegacyResponse(); - Fragment first = legacy.recv(); - Assert.assertNotNull(first); - Assert.assertEquals(1, legacy.calls); - - // and the default must keep the PREVIOUS behaviour, not merely link: it ignores the bound and - // defers to recv(), which is what such an implementation did before the overload existed - Fragment bounded = legacy.recv(5_000); - Assert.assertSame("the default must delegate to recv()", legacy.fragment, bounded); - Assert.assertEquals("and must not read anything of its own", 2, legacy.calls); - - Fragment unbounded = legacy.recv(0); - Assert.assertSame("a non-positive timeout is the legacy unbounded path, same delegation", - legacy.fragment, unbounded); - Assert.assertEquals(3, legacy.calls); - } - - private static void assertConstructorsPresent(Class type, String[] expectedSignatures) { - final Set actual = new TreeSet<>(); - for (Constructor c : type.getConstructors()) { - actual.add(signatureOf(c)); - } - final Set missing = new LinkedHashSet<>(); - for (String signature : expectedSignatures) { - if (!actual.contains(signature)) { - missing.add(signature); - } - } - Assert.assertTrue( - "these public " + type.getSimpleName() + " constructors are in the released jar and no longer " - + "exist, so a caller compiled against that release breaks with NoSuchMethodError. " - + "Add an overload instead of widening one in place.\n missing:\n " - + String.join("\n ", missing) + "\n present:\n " - + String.join("\n ", actual), - missing.isEmpty()); - } - - private static void assertSignaturesPresent(Class type, String methodName) { - final Set actual = new TreeSet<>(); - for (Method m : type.getMethods()) { - if (m.getName().equals(methodName)) { - actual.add(signatureOf(m)); - } - } - final Set missing = new LinkedHashSet<>(); - int expected = 0; - for (String signature : PRE_BRANCH_SIGNATURES) { - if (!signature.startsWith(methodName + "(")) { - continue; - } - expected++; - if (!actual.contains(signature)) { - missing.add(signature); - } - } - Assert.assertTrue("expected at least one pinned signature for " + methodName, expected > 0); - Assert.assertTrue( - "these " + type.getSimpleName() + '.' + methodName + " signatures existed at the merge base " - + "and no longer do, so a caller compiled against an earlier release breaks with " - + "NoSuchMethodError. Add an overload instead of retyping one.\n missing:\n " - + String.join("\n ", missing) + "\n present:\n " - + String.join("\n ", actual), - missing.isEmpty()); - } - - private static String signatureOf(Constructor c) { - final StringBuilder sb = new StringBuilder("("); - final Class[] params = c.getParameterTypes(); - for (int i = 0; i < params.length; i++) { - if (i > 0) { - sb.append(','); - } - sb.append(params[i].getName()); - } - return sb.append(')').toString(); - } - - private static String signatureOf(Method m) { - final StringBuilder sb = new StringBuilder(m.getName()).append('('); - final Class[] params = m.getParameterTypes(); - for (int i = 0; i < params.length; i++) { - if (i > 0) { - sb.append(','); - } - sb.append(params[i].getName()); - } - return sb.append(')').append(m.getReturnType().getName()).toString(); - } - - /** - * A {@link Response} written before {@code recv(int)} existed: it overrides {@code recv()} and nothing - * else. Its value is mostly at compile time - it does not compile against an abstract {@code recv(int)}. - */ - private static final class LegacyResponse implements Response { - private final Fragment fragment = new Fragment() { - @Override - public long hi() { - return 128L; - } - - @Override - public long lo() { - return 64L; - } - }; - private int calls; - - @Override - public Fragment recv() { - calls++; - return fragment; - } - } -} From 47a4322b91e8ea200b77f3c1001afb9d7061414b Mon Sep 17 00:00:00 2001 From: Jaromir Hamala Date: Thu, 15 Oct 2026 11:41:43 +0200 Subject: [PATCH 4/9] docs(qwp): durable-ack keepalive 0 no longer disables all PINGs --- core/src/main/java/io/questdb/client/Sender.java | 6 ++---- .../client/cutlass/qwp/client/QwpWebSocketSender.java | 4 ++-- .../qwp/client/sf/cursor/CursorWebSocketSendLoop.java | 6 +++--- 3 files changed, 7 insertions(+), 9 deletions(-) diff --git a/core/src/main/java/io/questdb/client/Sender.java b/core/src/main/java/io/questdb/client/Sender.java index 9113f2b8..f77a5634 100644 --- a/core/src/main/java/io/questdb/client/Sender.java +++ b/core/src/main/java/io/questdb/client/Sender.java @@ -2231,10 +2231,8 @@ public LineSenderBuilder drainOrphans(boolean enabled) { * the server periodically. Effective only when * {@link #requestDurableAck(boolean) request_durable_ack=on}. *

- * 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). + * Pass {@code 0} or negative to disable it. The idle PING + * ({@link #keepaliveIntervalMillis(long)}) is separate. *

* Default * {@value io.questdb.client.cutlass.qwp.client.sf.cursor.CursorWebSocketSendLoop#DEFAULT_DURABLE_ACK_KEEPALIVE_INTERVAL_MILLIS} 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 index 2892ef67..aedc4354 100644 --- 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 @@ -336,7 +336,7 @@ public class QwpWebSocketSender implements Sender { // 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. + // disables this PING (the idle keepalive PING is separate). private long durableAckKeepaliveIntervalMillis = CursorWebSocketSendLoop.DEFAULT_DURABLE_ACK_KEEPALIVE_INTERVAL_MILLIS; // Idle keepalive PING cadence for the I/O loop; 0 or negative disables it. @@ -890,7 +890,7 @@ public static QwpWebSocketSender connect( /** * 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 + * {@code 0} or negative disables this durable-ack PING (caller takes * responsibility for prodding the server). */ public static QwpWebSocketSender connect( 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 index a5256932..d46be0a1 100644 --- 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 @@ -134,7 +134,7 @@ public final class CursorWebSocketSendLoop implements QuietCloseable { * 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. + * server-side. {@code 0} or negative disables this PING, not the idle keepalive. */ public static final long DEFAULT_DURABLE_ACK_KEEPALIVE_INTERVAL_MILLIS = 200L; /** @@ -283,7 +283,7 @@ public final class CursorWebSocketSendLoop implements QuietCloseable { @TestOnly public static volatile boolean forceMirrorSeedFailureForTest; // Pre-converted to nanos for the comparison in sendPingIfDue. - // Zero or negative disables the keepalive entirely. + // Zero or negative disables the durable-ack PING only. private final long durableAckKeepaliveIntervalNanos; // Idle keepalive PING cadence in nanos; 0 or negative disables. Set before start(). private long keepaliveIntervalNanos = TimeUnit.MILLISECONDS.toNanos(DEFAULT_KEEPALIVE_INTERVAL_MILLIS); @@ -677,7 +677,7 @@ public CursorWebSocketSendLoop(WebSocketClient client, CursorSendEngine engine, /** * 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 + * Pass {@code 0} or negative to disable this durable-ack PING (the * caller takes responsibility for prodding the server, e.g. by sending * data, or by accepting indefinite waits on idle connections). */ From ebb6a755b5023a1ce3b14093b990fae1a2bac286 Mon Sep 17 00:00:00 2001 From: Jaromir Hamala Date: Thu, 15 Oct 2026 11:43:54 +0200 Subject: [PATCH 5/9] test(qwp): bound idle PING rate and assert PINGs never cost a reconnect --- .../test/cutlass/qwp/client/KeepaliveTest.java | 12 ++++++++++-- 1 file changed, 10 insertions(+), 2 deletions(-) diff --git a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/KeepaliveTest.java b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/KeepaliveTest.java index 41c63636..bb6fa770 100644 --- a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/KeepaliveTest.java +++ b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/KeepaliveTest.java @@ -36,16 +36,24 @@ public class KeepaliveTest { @Test public void testIdleConnectionSendsPings() throws Exception { + final long intervalMillis = 100; AtomicInteger pings = new AtomicInteger(); + long startNanos = System.nanoTime(); try (TestWebSocketServer server = startServer(pings); - Sender ignored = Sender.fromConfig(config(server, 100))) { + Sender ignored = Sender.fromConfig(config(server, intervalMillis))) { // generous: exits as soon as 3 pings land, and Windows CI can stall for seconds long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(30); while (pings.get() < 3 && System.nanoTime() < deadline) { Thread.sleep(10); } + // a PING must not cost a reconnect + Assert.assertEquals(1, server.handshakeCount()); } - Assert.assertTrue("expected periodic pings on an idle connection, saw " + pings.get(), pings.get() >= 3); + long elapsedMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startNanos); + int seen = pings.get(); + Assert.assertTrue("expected periodic pings on an idle connection, saw " + seen, seen >= 3); + // the gate spaces PINGs at least one interval apart, so this holds however slow the box is + Assert.assertTrue("ping flood: " + seen + " in " + elapsedMillis + " ms", seen <= elapsedMillis / intervalMillis + 1); } @Test From 03b7fc705df3348c1798c3bd9ef81819bc04e1eb Mon Sep 17 00:00:00 2001 From: Jaromir Hamala Date: Thu, 15 Oct 2026 12:20:48 +0200 Subject: [PATCH 6/9] refactor(qwp): drop dead keepalive reset on durable-ack reconnect pendingDurable is emptied on reconnect and refills only after a frame is sent on the new connection, which stamps lastFrameOrPingNanos, so the reset could never make the durable-ack PING fire sooner. --- .../qwp/client/sf/cursor/CursorWebSocketSendLoop.java | 8 +------- 1 file changed, 1 insertion(+), 7 deletions(-) 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 index d46be0a1..f73c5de5 100644 --- 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 @@ -526,8 +526,7 @@ public final class CursorWebSocketSendLoop implements QuietCloseable { // (trySendOne) or a keepalive PING (sendPingIfDue). // Throttles the keepalive PINGs so they fire only when the // configured interval has elapsed since the most recent outbound event. - // Set when the I/O loop starts; durable-ack mode backdates it on - // reconnect so its keepalive fires at once. + // Set when the I/O loop starts. private long lastFrameOrPingNanos; // Throttles the catch-up oversize WARN. Only consulted when the server // advertises NO batch cap, where the loop has no terminal to fall back on @@ -1918,11 +1917,6 @@ private void clearDurableAckTracking() { releasePendingEntry(pendingDurable.pollFirst()); } durableTableWatermarks.clear(); - // Backdate the keepalive throttle by one durable interval so the new - // connection can prod the server immediately rather than waiting out - // the leftover interval from before the reconnect. Relative to now, - // not 0, so the far longer idle keepalive is not also made due. - lastFrameOrPingNanos = System.nanoTime() - durableAckKeepaliveIntervalNanos; } private static boolean isCatchUpCapGap(Throwable t) { From 31f9e41f88d5a546727f71a18e22a051c41e3a1a Mon Sep 17 00:00:00 2001 From: Jaromir Hamala Date: Thu, 15 Oct 2026 13:11:18 +0200 Subject: [PATCH 7/9] fix(qwp): send keepalive PINGs with the data-frame timeout A PING timeout reconnects and replays unacked frames. With 1 s, a GC pause or a slow-draining server could tear down a live connection. --- .../client/cutlass/http/client/WebSocketClient.java | 6 +++--- .../qwp/client/sf/cursor/CursorWebSocketSendLoop.java | 7 +++---- .../test/cutlass/http/client/WebSocketClientTest.java | 2 +- 3 files changed, 7 insertions(+), 8 deletions(-) diff --git a/core/src/main/java/io/questdb/client/cutlass/http/client/WebSocketClient.java b/core/src/main/java/io/questdb/client/cutlass/http/client/WebSocketClient.java index 1290fcab..8ab593d5 100644 --- a/core/src/main/java/io/questdb/client/cutlass/http/client/WebSocketClient.java +++ b/core/src/main/java/io/questdb/client/cutlass/http/client/WebSocketClient.java @@ -526,14 +526,14 @@ public void sendCloseFrame(int code, String reason, int timeout) { } /** - * Sends a ping frame. + * Sends a ping frame with the same timeout as data frames. */ - public void sendPing(int timeout) { + public void sendPing() { checkConnected(); controlFrameBuffer.reset(); WebSocketSendBuffer.FrameInfo frame = controlFrameBuffer.writePingFrame(); try { - doSend(controlFrameBuffer.getBufferPtr() + frame.offset, frame.length, timeout); + doSend(controlFrameBuffer.getBufferPtr() + frame.offset, frame.length, defaultTimeout); } finally { controlFrameBuffer.reset(); } 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 index f73c5de5..c58f918f 100644 --- 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 @@ -2868,10 +2868,9 @@ private void sendPingIfDue(long intervalNanos) { } lastFrameOrPingNanos = now; try { - // 1 s, not the 600 s data-frame timeout: a socket that cannot take a - // 6-byte PING after an idle interval belongs to a wedged server, and - // reconnect + replay beats blocking the I/O thread for minutes. - client.sendPing(1000); + // A timeout here reconnects and replays unacked frames, so it must be as + // tolerant as a data send: a slow-draining server or a GC pause is not a dead peer. + client.sendPing(); } catch (Throwable t) { fail(t); } diff --git a/core/src/test/java/io/questdb/client/test/cutlass/http/client/WebSocketClientTest.java b/core/src/test/java/io/questdb/client/test/cutlass/http/client/WebSocketClientTest.java index 93a380f8..f1d690b9 100644 --- a/core/src/test/java/io/questdb/client/test/cutlass/http/client/WebSocketClientTest.java +++ b/core/src/test/java/io/questdb/client/test/cutlass/http/client/WebSocketClientTest.java @@ -394,7 +394,7 @@ public void testSendPingDoesNotClobberSendBuffer() throws Exception { // sendPing() should use controlFrameBuffer, not sendBuffer try { - client.sendPing(1000); + client.sendPing(); } catch (HttpClientException ignored) { // Expected: doSend() fails because there's no real socket } From d673d7507e8822639ee585fee7eeefb74a135a2e Mon Sep 17 00:00:00 2001 From: Jaromir Hamala Date: Tue, 20 Oct 2026 16:23:12 +0200 Subject: [PATCH 8/9] feat(qwp): make idle keepalive PING opt-in --- core/src/main/java/io/questdb/client/Sender.java | 8 ++++---- .../qwp/client/sf/cursor/CursorWebSocketSendLoop.java | 4 ++-- .../client/test/cutlass/qwp/client/KeepaliveTest.java | 4 ++-- 3 files changed, 8 insertions(+), 8 deletions(-) diff --git a/core/src/main/java/io/questdb/client/Sender.java b/core/src/main/java/io/questdb/client/Sender.java index f77a5634..62138a55 100644 --- a/core/src/main/java/io/questdb/client/Sender.java +++ b/core/src/main/java/io/questdb/client/Sender.java @@ -2607,10 +2607,10 @@ public LineSenderBuilder initialConnectRetry(boolean enabled) { * server's own idle timeout, from closing a quiet connection, which * would otherwise cost a reconnect each time. *

- * Pass {@code 0} or negative to disable it. Default - * {@value io.questdb.client.cutlass.qwp.client.sf.cursor.CursorWebSocketSendLoop#DEFAULT_KEEPALIVE_INTERVAL_MILLIS} - * ms. Background orphan drainers ({@link #drainOrphans(boolean)}) - * always use the default. WebSocket transport only. + * Off by default; opt in with a positive value ({@code 30000} suits + * an AWS ALB). {@code 0} or negative disables it. Background orphan + * drainers ({@link #drainOrphans(boolean)}) never send it. WebSocket + * transport only. */ public LineSenderBuilder keepaliveIntervalMillis(long millis) { if (protocol != PARAMETER_NOT_SET_EXPLICITLY && protocol != PROTOCOL_WEBSOCKET) { 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 index c58f918f..68803fbc 100644 --- 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 @@ -145,9 +145,9 @@ public final class CursorWebSocketSendLoop implements QuietCloseable { * server's own idle reaper ({@code http.net.connection.timeout}, 5 min), * and the loop pays a reconnect each time. TCP keepalive does not help * through an L7 proxy, which terminates TCP. {@code 0} or negative - * disables it. + * disables it. Opt-in: off by default, {@code 30000} suits an AWS ALB. */ - public static final long DEFAULT_KEEPALIVE_INTERVAL_MILLIS = 30_000L; + public static final long DEFAULT_KEEPALIVE_INTERVAL_MILLIS = 0L; public static final long DEFAULT_PARK_NANOS = 50_000L; // 50us idle backoff /** * Default initial reconnect backoff (100 ms). diff --git a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/KeepaliveTest.java b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/KeepaliveTest.java index bb6fa770..91e52291 100644 --- a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/KeepaliveTest.java +++ b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/KeepaliveTest.java @@ -57,10 +57,10 @@ public void testIdleConnectionSendsPings() throws Exception { } @Test - public void testZeroIntervalDisablesPings() throws Exception { + public void testNoPingsByDefault() throws Exception { AtomicInteger pings = new AtomicInteger(); try (TestWebSocketServer server = startServer(pings); - Sender ignored = Sender.fromConfig(config(server, 0))) { + Sender ignored = Sender.fromConfig("ws::addr=localhost:" + server.getPort() + ";")) { // A broken disable (interval 0 treated as "always due") pings on every idle pass Thread.sleep(500); } From e600782d5e5c38d2e313de7ee8f2dd481a9a2a83 Mon Sep 17 00:00:00 2001 From: Jaromir Hamala Date: Tue, 20 Oct 2026 17:46:53 +0200 Subject: [PATCH 9/9] fix(qwp): drop replay count from reconnect log The count only covered frames sent on an earlier connection of the same loop, so a sender resuming a recovered SF slot logged "0 unacked frames to replay" right before re-sending its whole backlog. Log just the resume FSN; getTotalFramesReplayed() still reports re-sends. --- .../client/sf/cursor/CursorWebSocketSendLoop.java | 14 ++++---------- 1 file changed, 4 insertions(+), 10 deletions(-) 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 index 68803fbc..598f387c 100644 --- 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 @@ -2050,14 +2050,11 @@ private void connectWalk(Throwable initial, String phase, long paceFirstAttemptM } break; } - long resumeFsn = swapClient(newClient); + swapClient(newClient); counters.reconnects.incrementAndGet(); long elapsedMs = (System.nanoTime() - outageStartNanos) / 1_000_000L; - // swapClient cleared replayTargetFsn when nothing below it is unacked - long replayFrames = replayTargetFsn >= 0L ? replayTargetFsn - resumeFsn + 1L : 0L; - LOG.info("cursor I/O loop {} succeeded after {}ms, {} attempts; " - + "resuming at FSN {}, {} unacked frames to replay", - phase, elapsedMs, attempts, externalFsnBase + resumeFsn, replayFrames); + LOG.info("cursor I/O loop {} succeeded after {}ms, {} attempts; resuming at FSN {}", + phase, elapsedMs, attempts, externalFsnBase + engine.ackedFsn() + 1L); return; } // A null factory result is an unsuccessful connect state, not a cap-gap @@ -2898,10 +2895,8 @@ private void snapshotReplayTarget() { * realign {@code fsnAtZero} to the next unacked FSN, restart wire * sequencing from 0, and reposition the cursor so the next * {@link #trySendOne} call replays the first unacked frame. - * - * @return the first unacked FSN, where sending resumes on the new connection */ - private long swapClient(WebSocketClient newClient) { + private void swapClient(WebSocketClient newClient) { WebSocketClient old = this.client; this.client = newClient; this.lastReconnectError = null; @@ -2945,7 +2940,6 @@ private long swapClient(WebSocketClient newClient) { // established, so the sender stays INITIALIZING and a subsequent // endpoint-policy rejection still latches the startup terminal. this.hasEverConnected = true; - return replayStart; } /**