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 9 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
9 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
c329caa
Merge branch 'main' into nw_adaptive_commit
eugenels Sep 14, 2026
5b10a31
Merge remote-tracking branch 'origin/main' into nw_adaptive_commit
eugenels Oct 12, 2026
bf373ac
Name tier set and echo in durable-ack mismatch
eugenels Oct 18, 2026
ea7aa1a
Test orphan drainer with the local ack tier
eugenels Oct 18, 2026
9d9aad5
Merge remote-tracking branch 'origin/main' into nw_adaptive_commit
eugenels Oct 22, 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 @@ -36,6 +36,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 @@ -1227,7 +1228,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 @@ -2693,20 +2694,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 @@ -3695,7 +3736,7 @@ private QwpWebSocketSender buildWebSocket(HttpTokenProvider tokenProvider) {
actualAutoFlushBytes,
actualAutoFlushIntervalNanos,
wsAuthHeader,
requestDurableAck,
durableAckTiers,
cursorEngine,
actualCloseFlushTimeoutMillis,
actualReconnectMaxDurationMillis,
Expand Down Expand Up @@ -4162,13 +4203,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 @@ -4636,13 +4675,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 @@ -4772,7 +4809,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 @@ -4892,7 +4929,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
85 changes: 56 additions & 29 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 @@ -136,18 +136,25 @@ public abstract class WebSocketClient implements QuietCloseable {
// down to its own hard limit.
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;
// Requested durable-ack tier set (DurableAckTiers bitmask), sent as
// X-QWP-Request-Durable-Ack: <DurableAckTiers.requestHeaderValue>
private int qwpDurableAckTiers = DurableAckTiers.NONE;
// Receive buffer (native memory)
private long recvBufPtr;
private int recvBufSize;
private int recvPos; // Write position
private int recvReadPos; // Read position
// Raw X-QWP-Durable-Ack value from the 101 upgrade response, or null when
// the server sent no such header. Reported in the durable-ack mismatch
// error so an operator sees what the server answered.
private String serverDurableAckEcho;
// Set during upgrade response validation when the server echoed
// X-QWP-Durable-Ack: enabled. Tells the sender it landed on a server that
// will actually emit STATUS_DURABLE_ACK frames, so its store-and-forward
// path can rely on durable-ack-driven trim. Absence (after opting in via
// setQwpRequestDurableAck) is the early-fail signal.
// X-QWP-Durable-Ack with the token DurableAckTiers.expectedConfirmToken
// expects for the requested set (the set itself, or "enabled" for a
// legacy request). Tells the sender it landed on a server that will emit
// the requested ack frames, so its store-and-forward path can rely on
// durable-ack-driven trim. Absence (after requesting tiers via
// setQwpDurableAckTiers) is the early-fail signal.
private boolean serverDurableAckEnabled;
// Server's hard cap on ingest QWP message payload bytes, extracted from
// X-QWP-Max-Batch-Size on the 101 upgrade response. 0 when the server did
Expand Down Expand Up @@ -334,6 +341,15 @@ public WebSocketSendBuffer getSendBuffer() {
return sendBuffer;
}

/**
* Raw {@code X-QWP-Durable-Ack} value from the most recent 101 upgrade
* response, trimmed, or null when the server sent no such header.
* Meaningful only after {@link #upgrade} returns.
*/
public String getServerDurableAckEcho() {
return serverDurableAckEcho;
}

/**
* Server-advertised hard cap on QWP ingest payload bytes, taken from the
* {@code X-QWP-Max-Batch-Size} response header on the 101 upgrade. Returns
Expand Down Expand Up @@ -414,10 +430,13 @@ public boolean isConnected() {
}

/**
* Returns true when the server echoed X-QWP-Durable-Ack: enabled in the
* 101 upgrade response. Meaningful only after {@link #upgrade} returns;
* always false when the client did not opt in via
* {@link #setQwpRequestDurableAck}.
* Returns true when the server granted the requested durable-ack tier set:
* the 101 upgrade response carried {@code X-QWP-Durable-Ack} with the
* token {@link DurableAckTiers#expectedConfirmToken} expects (the
* requested set, such as {@code local} or {@code local,replicated}, or
* {@code enabled} for a legacy request). Meaningful only after
* {@link #upgrade} returns; always false when no tier was requested via
* {@link #setQwpDurableAckTiers}.
*/
public boolean isServerDurableAckEnabled() {
return serverDurableAckEnabled;
Expand Down Expand Up @@ -599,13 +618,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 @@ -708,8 +728,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 @@ -812,7 +834,7 @@ private static int extractContentEncodingZstdLevel(String response) {
return 0;
}

private static boolean extractDurableAckEnabled(String response) {
private static String extractDurableAckEcho(String response) {
int headerLen = QWP_DURABLE_ACK_HEADER_NAME.length();
int responseLen = response.length();
for (int i = 0; i <= responseLen - headerLen; i++) {
Expand All @@ -822,11 +844,10 @@ private static boolean extractDurableAckEnabled(String response) {
if (lineEnd < 0) {
lineEnd = responseLen;
}
String value = response.substring(valueStart, lineEnd).trim();
return value.equalsIgnoreCase(QWP_DURABLE_ACK_ENABLED_VALUE);
return response.substring(valueStart, lineEnd).trim();
}
}
return false;
return null;
}

private static int extractMaxBatchSize(String response) {
Expand Down Expand Up @@ -1430,12 +1451,18 @@ private void validateUpgradeResponse(int headerEnd) {
throw new QwpVersionMismatchException(serverQwpVersion, qwpMaxVersion);
}

// Extract X-QWP-Durable-Ack confirmation (optional, absent on servers
// without primary replication or when the client did not opt in).
// Only meaningful when qwpRequestDurableAck is true; the sender
// checks this value to fail at connect rather than silently
// missing trim signals.
serverDurableAckEnabled = extractDurableAckEnabled(response);
// Extract X-QWP-Durable-Ack confirmation (optional, absent when the
// server cannot grant the requested tier set or the client requested
// none). Only meaningful when qwpDurableAckTiers is not NONE; the
// sender checks this value to fail at connect rather than silently
// missing trim signals. 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.
serverDurableAckEcho = extractDurableAckEcho(response);
String expectedDurableAckToken = DurableAckTiers.expectedConfirmToken(qwpDurableAckTiers);
serverDurableAckEnabled = expectedDurableAckToken != null
&& expectedDurableAckToken.equalsIgnoreCase(serverDurableAckEcho);

// 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