Skip to content

feat(core): add local durable acknowledgements to the QWP sender #98

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
eugenels wants to merge 4 commits into
base: main
Choose a base branch
from
Open

feat(core): add local durable acknowledgements to the QWP sender #98

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
7a05cb6
Add local durable ack tier to QWP sender
eugenels Sep 11, 2026
2babfc8
Restore boolean connect overloads for binary compat
eugenels Sep 11, 2026
cf2db9a
Expose the local-fsync watermark per table
eugenels Sep 11, 2026
ea04dc1
Cover durable-ack tiers with real tests
eugenels Sep 11, 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
77 changes: 57 additions & 20 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 @@ -34,6 +34,7 @@
import io.questdb.client.cutlass.line.http.AbstractLineHttpSender;
import io.questdb.client.cutlass.line.tcp.DelegatingTlsChannel;
import io.questdb.client.cutlass.line.tcp.PlainTcpLineChannel;
import io.questdb.client.cutlass.qwp.client.DurableAckTiers;
import io.questdb.client.cutlass.qwp.client.QwpUdpSender;
import io.questdb.client.cutlass.qwp.client.QwpWebSocketSender;
import io.questdb.client.cutlass.qwp.client.sf.cursor.CursorSendEngine;
Expand Down Expand Up @@ -1145,7 +1146,7 @@ public int getConnectTimeout() {
// max backoff (default 5_000) for the cursor I/O loop's exponential
// retry-with-jitter loop.
private long reconnectMaxDurationMillis = PARAMETER_NOT_SET_EXPLICITLY;
private boolean requestDurableAck;
private int durableAckTiers = DurableAckTiers.NONE;
private int retryTimeoutMillis = PARAMETER_NOT_SET_EXPLICITLY;
private boolean transactional;
private String senderId = DEFAULT_SENDER_ID;
Expand Down Expand Up @@ -1698,7 +1699,7 @@ public Sender build() {
actualAutoFlushBytes,
actualAutoFlushIntervalNanos,
wsAuthHeader,
requestDurableAck,
durableAckTiers,
cursorEngine,
actualCloseFlushTimeoutMillis,
actualReconnectMaxDurationMillis,
Expand Down Expand Up @@ -2779,20 +2780,60 @@ public LineSenderBuilder reconnectMaxDurationMillis(long millis) {
}

/**
* Opts the connection in for STATUS_DURABLE_ACK frames. When enabled,
* servers with primary replication will emit per-table durable-upload
* watermarks as WAL data reaches the object store.
* Opts the connection in for STATUS_DURABLE_ACK frames, using the
* legacy "true" request token. Equivalent to
* {@code requestDurableAck("on")}: requests the replicated tier, so
* servers without primary replication deny the request and the
* sender fails at connect.
* <p>
* This setting is only supported for WebSocket transport.
*
* @param enabled true to request durable ACKs
* @return this instance for method chaining
*/
public LineSenderBuilder requestDurableAck(boolean enabled) {
return requestDurableAckTiers(enabled
? DurableAckTiers.REPLICATED | DurableAckTiers.LEGACY_TRUE
: DurableAckTiers.NONE);
}

/**
* Requests durable-ack streams by tier set. Accepted values:
* {@code off}, {@code on} (legacy alias for the replicated tier),
* {@code local}, {@code replicated}, {@code local,replicated}.
* <ul>
* <li>{@code local} -- the server emits STATUS_LOCAL_DURABLE_ACK
* frames once commits are fdatasync-durable on its disk; the
* sender trims its store-and-forward copy on them.</li>
* <li>{@code replicated} -- the server emits STATUS_DURABLE_ACK
* frames once commits reach the object store; the sender trims
* on them.</li>
* <li>{@code local,replicated} -- both streams; the sender trims on
* the replicated ack (the strongest requested guarantee) and
* receives local acks as early progress signals.</li>
* </ul>
* The server grants the full requested set or denies the request
* entirely (the sender then fails at connect); it never substitutes
* a weaker guarantee.
* <p>
* This setting is only supported for WebSocket transport.
*
* @param tiers the requested tier set
* @return this instance for method chaining
*/
public LineSenderBuilder requestDurableAck(CharSequence tiers) {
int parsed = DurableAckTiers.parseConfigValue(tiers);
if (parsed < 0) {
throw new LineSenderException("invalid request_durable_ack [value=").put(tiers).put(", allowed-values=[on, off, local, replicated, local,replicated]]");
}
return requestDurableAckTiers(parsed);
}

private LineSenderBuilder requestDurableAckTiers(int tiers) {
if (protocol != PARAMETER_NOT_SET_EXPLICITLY && protocol != PROTOCOL_WEBSOCKET) {
throw new LineSenderException("request_durable_ack is only supported for WebSocket transport");
}
this.requestDurableAck = enabled;
this.durableAckTiers = tiers;
return this;
}

Expand Down Expand Up @@ -3762,13 +3803,11 @@ private LineSenderBuilder fromConfig(CharSequence configurationString) {
throw new LineSenderException("request_durable_ack is only supported for WebSocket transport");
}
pos = getValue(configurationString, pos, sink, "request_durable_ack");
if (Chars.equalsIgnoreCase("on", sink)) {
requestDurableAck(true);
} else if (Chars.equalsIgnoreCase("off", sink)) {
requestDurableAck(false);
} else {
throw new LineSenderException("invalid request_durable_ack [value=").put(sink).put(", allowed-values=[on, off]]");
int tiers = DurableAckTiers.parseConfigValue(sink);
if (tiers < 0) {
throw new LineSenderException("invalid request_durable_ack [value=").put(sink).put(", allowed-values=[on, off, local, replicated, local,replicated]]");
}
requestDurableAckTiers(tiers);
} else if (Chars.equals("transaction", sink)) {
if (protocol != PROTOCOL_WEBSOCKET) {
throw new LineSenderException("transaction is only supported for WebSocket transport");
Expand Down Expand Up @@ -4180,13 +4219,11 @@ private LineSenderBuilder fromConfigWebSocket(CharSequence configurationString)
}
s = view.getStr("request_durable_ack");
if (s != null) {
if (s.equalsIgnoreCase("on")) {
requestDurableAck(true);
} else if (s.equalsIgnoreCase("off")) {
requestDurableAck(false);
} else {
throw new LineSenderException("invalid request_durable_ack [value=").put(s).put(", allowed-values=[on, off]]");
int tiers = DurableAckTiers.parseConfigValue(s);
if (tiers < 0) {
throw new LineSenderException("invalid request_durable_ack [value=").put(s).put(", allowed-values=[on, off, local, replicated, local,replicated]]");
}
requestDurableAckTiers(tiers);
}
s = view.getStr("drain_orphans");
if (s != null) {
Expand Down Expand Up @@ -4306,7 +4343,7 @@ public java.util.Map<String, Object> wsConfigSnapshotForTest() {
m.put("auto_flush_interval", autoFlushIntervalMillis);
m.put("max_name_len", maxNameLength);
m.put("transaction", transactional);
m.put("request_durable_ack", requestDurableAck);
m.put("request_durable_ack", DurableAckTiers.configValue(durableAckTiers));
m.put("sender_id", senderId);
m.put("sf_dir", sfDir);
m.put("sf_max_segment_bytes", sfMaxSegmentBytes);
Expand Down Expand Up @@ -4400,7 +4437,7 @@ private void validateParameters() {
.put(", requestedCapacity=").put(bufferCapacity)
.put("]");
}
if (requestDurableAck && protocol != PROTOCOL_WEBSOCKET) {
if (durableAckTiers != DurableAckTiers.NONE && protocol != PROTOCOL_WEBSOCKET) {
throw new LineSenderException("request_durable_ack is only supported for WebSocket transport");
}
if (protocol == PROTOCOL_HTTP) {
Expand Down
37 changes: 24 additions & 13 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 @@ -25,6 +25,7 @@
package io.questdb.client.cutlass.http.client;

import io.questdb.client.HttpClientConfiguration;
import io.questdb.client.cutlass.qwp.client.DurableAckTiers;
import io.questdb.client.cutlass.qwp.client.QwpVersionMismatchException;
import io.questdb.client.cutlass.qwp.websocket.WebSocketCloseCode;
import io.questdb.client.cutlass.qwp.websocket.WebSocketFrameParser;
Expand Down Expand Up @@ -80,7 +81,6 @@ public abstract class WebSocketClient implements QuietCloseable {
private static final String QUESTDB_ROLE_HEADER_NAME = "X-QuestDB-Role:";
private static final String QUESTDB_ZONE_HEADER_NAME = "X-QuestDB-Zone:";
private static final String QWP_CONTENT_ENCODING_HEADER_NAME = "X-QWP-Content-Encoding:";
private static final String QWP_DURABLE_ACK_ENABLED_VALUE = "enabled";
private static final String QWP_DURABLE_ACK_HEADER_NAME = "X-QWP-Durable-Ack:";
private static final String QWP_MAX_BATCH_SIZE_HEADER_NAME = "X-QWP-Max-Batch-Size:";
private static final String QWP_VERSION_HEADER_NAME = "X-QWP-Version:";
Expand Down Expand Up @@ -137,7 +137,7 @@ public abstract class WebSocketClient implements QuietCloseable {
private int qwpMaxBatchRows;
private int qwpMaxVersion = 1;
// Opt-in for STATUS_DURABLE_ACK frames; sent as X-QWP-Request-Durable-Ack: true
private boolean qwpRequestDurableAck;
private int qwpDurableAckTiers = DurableAckTiers.NONE;
// Receive buffer (native memory)
private long recvBufPtr;
private int recvBufSize;
Expand Down Expand Up @@ -586,13 +586,14 @@ public void setQwpMaxVersion(int maxVersion) {
}

/**
* Enables the opt-in X-QWP-Request-Durable-Ack upgrade header. When set,
* servers with primary replication configured will additionally emit
* STATUS_DURABLE_ACK frames as the WAL containing committed client
* messages reaches the object store.
* Sets the requested durable-ack tier set ({@link DurableAckTiers}
* bitmask) for the opt-in X-QWP-Request-Durable-Ack upgrade header. When
* granted, the server emits STATUS_DURABLE_ACK frames (replicated tier)
* and/or STATUS_LOCAL_DURABLE_ACK frames (local tier) as commits reach
* the corresponding durability frontier.
*/
public void setQwpRequestDurableAck(boolean enabled) {
this.qwpRequestDurableAck = enabled;
public void setQwpDurableAckTiers(int tiers) {
this.qwpDurableAckTiers = tiers;
}

/**
Expand Down Expand Up @@ -694,8 +695,10 @@ public void upgrade(CharSequence path, int timeout, CharSequence authorizationHe
sendBuffer.putAscii(Integer.toString(qwpMaxBatchRows));
sendBuffer.putAscii("\r\n");
}
if (qwpRequestDurableAck) {
sendBuffer.putAscii("X-QWP-Request-Durable-Ack: true\r\n");
if (qwpDurableAckTiers != DurableAckTiers.NONE) {
sendBuffer.putAscii("X-QWP-Request-Durable-Ack: ");
sendBuffer.putAscii(DurableAckTiers.requestHeaderValue(qwpDurableAckTiers));
sendBuffer.putAscii("\r\n");
}
if (authorizationHeader != null) {
sendBuffer.putAscii("Authorization: ");
Expand Down Expand Up @@ -798,7 +801,10 @@ private static int extractContentEncodingZstdLevel(String response) {
return 0;
}

private static boolean extractDurableAckEnabled(String response) {
private static boolean extractDurableAckConfirmed(String response, String expectedToken) {
if (expectedToken == null) {
return false;
}
int headerLen = QWP_DURABLE_ACK_HEADER_NAME.length();
int responseLen = response.length();
for (int i = 0; i <= responseLen - headerLen; i++) {
Expand All @@ -809,7 +815,11 @@ private static boolean extractDurableAckEnabled(String response) {
lineEnd = responseLen;
}
String value = response.substring(valueStart, lineEnd).trim();
return value.equalsIgnoreCase(QWP_DURABLE_ACK_ENABLED_VALUE);
// The server echoes the granted set verbatim (or the
// "enabled" token for a legacy "true" request); anything
// else is a partial or foreign grant and counts as a
// denial -- all-or-nothing, never a silent downgrade.
return value.equalsIgnoreCase(expectedToken);
}
}
return false;
Expand Down Expand Up @@ -1390,7 +1400,8 @@ private void validateUpgradeResponse(int headerEnd) {
// Only meaningful when qwpRequestDurableAck is true; the sender
// checks this value to fail at connect rather than silently
// missing trim signals.
serverDurableAckEnabled = extractDurableAckEnabled(response);
serverDurableAckEnabled = extractDurableAckConfirmed(
response, DurableAckTiers.expectedConfirmToken(qwpDurableAckTiers));

// Extract X-QWP-Max-Batch-Size (optional). Older servers omit it; the
// sender falls back to its locally configured byte budget in that case.
Expand Down
Loading
Loading