Skip to content

feat(qwp): keep idle WebSocket connections alive with PINGs #109

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
jerrinot wants to merge 7 commits into
base: main
Choose a base branch
from
Open

feat(qwp): keep idle WebSocket connections alive with PINGs #109

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
352361e
fix(qwp): log the real resume FSN and replay count on reconnect
jerrinot Oct 15, 2026
37e2ff2
feat(qwp): send keepalive PINGs on idle connections
jerrinot Oct 15, 2026
c96375f
refactor(qwp): drop keepalive factory overload and ExportedApiCompati…
jerrinot Oct 15, 2026
47a4322
docs(qwp): durable-ack keepalive 0 no longer disables all PINGs
jerrinot Oct 15, 2026
ebb6a75
test(qwp): bound idle PING rate and assert PINGs never cost a reconnect
jerrinot Oct 15, 2026
03b7fc7
refactor(qwp): drop dead keepalive reset on durable-ack reconnect
jerrinot Oct 15, 2026
31f9e41
fix(qwp): send keepalive PINGs with the data-frame timeout
jerrinot Oct 15, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
41 changes: 36 additions & 5 deletions core/src/main/java/io/questdb/client/Sender.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -2229,10 +2231,8 @@ public LineSenderBuilder drainOrphans(boolean enabled) {
* 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).
* Pass {@code 0} or negative to disable it. The idle PING
* ({@link #keepaliveIntervalMillis(long)}) is separate.
* <p>
* Default
* {@value io.questdb.client.cutlass.qwp.client.sf.cursor.CursorWebSocketSendLoop#DEFAULT_DURABLE_ACK_KEEPALIVE_INTERVAL_MILLIS}
Expand Down Expand Up @@ -2599,6 +2599,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.
* <p>
* 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
Expand Down Expand Up @@ -4157,6 +4178,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");
Expand Down Expand Up @@ -4468,6 +4495,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"));
}
Expand Down Expand Up @@ -4677,6 +4707,7 @@ public java.util.Map<String, Object> 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);
Expand Down
6 changes: 3 additions & 3 deletions core/src/main/java/io/questdb/client/cutlass/http/client/WebSocketClient.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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();
}
Expand Down
27 changes: 18 additions & 9 deletions core/src/main/java/io/questdb/client/cutlass/qwp/client/QwpWebSocketSender.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -336,9 +336,11 @@ 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.
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
Expand Down Expand Up @@ -888,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(
Expand Down Expand Up @@ -1047,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);
}

/**
Expand Down Expand Up @@ -1105,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);
}

/**
Expand Down Expand Up @@ -1166,7 +1170,7 @@ autoFlushIntervalNanos, fixedAuthHeader(authorizationHeader),
maxFrameRejections, poisonMinEscalationWindowMillis,
catchUpCapGapMinEscalationWindowMillis,
symbolDictResetEnabled, symbolDictResetThresholdSymbols,
symbolDictResetMaxWaitMillis);
symbolDictResetMaxWaitMillis, CursorWebSocketSendLoop.DEFAULT_KEEPALIVE_INTERVAL_MILLIS);
}

/**
Expand All @@ -1176,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.
* <p>
* Recycling requires a non-null engine rebuild factory installed via
* {@link #setEngineRebuildFactory(EngineRebuildFactory)};
Expand Down Expand Up @@ -1219,7 +1223,8 @@ 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);
}

/**
Expand All @@ -1228,7 +1233,8 @@ public static QwpWebSocketSender connectWithCredentialSupplier(
* 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}).
* <p>
* Named apart from {@code connect} for the reason given on
* {@link #connectWithCredentialSupplier(List, ClientTlsConfiguration, int, int, long, Supplier,
Expand Down Expand Up @@ -1272,7 +1278,8 @@ public static QwpWebSocketSender connectWithCredentialSupplier(
long catchUpCapGapMinEscalationWindowMillis,
boolean symbolDictResetEnabled,
int symbolDictResetThresholdSymbols,
long symbolDictResetMaxWaitMillis
long symbolDictResetMaxWaitMillis,
long keepaliveIntervalMillis
) {
if (symbolDictResetThresholdSymbols <= 0
|| symbolDictResetThresholdSymbols > QwpConstants.MAX_SYMBOL_DICTIONARY_SIZE / 2) {
Expand Down Expand Up @@ -1301,6 +1308,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;
Expand Down Expand Up @@ -4896,6 +4904,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
Expand Down
Loading
Loading