* This setting is only supported for WebSocket transport. * @@ -2789,10 +2792,48 @@ public LineSenderBuilder reconnectMaxDurationMillis(long millis) { * @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}. + *
+ * 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;
}
@@ -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");
@@ -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) {
@@ -4306,7 +4343,7 @@ public java.util.Map
+ * The server grants the full requested set or denies the request entirely
+ * (no confirmation header); it never substitutes a weaker guarantee. The
+ * sender trims its store-and-forward copy on the strongest requested tier's
+ * ack; with both tiers requested, local acks arrive as progress signals only.
+ */
+public final class DurableAckTiers {
+
+ public static final int NONE = 0;
+ public static final int LOCAL = 1;
+ public static final int REPLICATED = 2;
+ // Modifier bit, only ever combined with REPLICATED: send the legacy "true"
+ // request token and expect the historical "enabled" confirmation.
+ public static final int LEGACY_TRUE = 4;
+
+ private DurableAckTiers() {
+ }
+
+ /**
+ * The config-string form of a tier set, the inverse of
+ * {@link #parseConfigValue(CharSequence)} (legacy sets print as "on").
+ */
+ public static String configValue(int tiers) {
+ if ((tiers & LEGACY_TRUE) != 0) {
+ return "on";
+ }
+ switch (tiers & (LOCAL | REPLICATED)) {
+ case LOCAL:
+ return "local";
+ case REPLICATED:
+ return "replicated";
+ case LOCAL | REPLICATED:
+ return "local,replicated";
+ default:
+ return "off";
+ }
+ }
+
+ /**
+ * The X-QWP-Durable-Ack confirmation token the server must echo for this
+ * request, or null when no tier is requested. A legacy request expects
+ * the historical {@code "enabled"} token; explicit requests expect their
+ * own token set back verbatim.
+ */
+ public static String expectedConfirmToken(int tiers) {
+ if ((tiers & LEGACY_TRUE) != 0) {
+ return "enabled";
+ }
+ return explicitToken(tiers);
+ }
+
+ public static boolean hasLocal(int tiers) {
+ return (tiers & LOCAL) != 0;
+ }
+
+ public static boolean hasReplicated(int tiers) {
+ return (tiers & REPLICATED) != 0;
+ }
+
+ /**
+ * True when the sender's trim trigger is {@code STATUS_LOCAL_DURABLE_ACK}:
+ * the local tier is requested without the replicated one. Any request
+ * including the replicated tier trims on {@code STATUS_DURABLE_ACK} — the
+ * strongest requested guarantee wins.
+ */
+ public static boolean isTrimOnLocalAck(int tiers) {
+ return hasLocal(tiers) && !hasReplicated(tiers);
+ }
+
+ /**
+ * Parses a {@code request_durable_ack} value into a tier set, or -1 for
+ * an unrecognized value. {@code on} is the legacy alias for the
+ * replicated tier (its shipped meaning); {@code off} is {@link #NONE}.
+ */
+ public static int parseConfigValue(CharSequence value) {
+ if (value == null) {
+ return -1;
+ }
+ String v = value.toString().trim();
+ if (v.equalsIgnoreCase("off")) {
+ return NONE;
+ }
+ if (v.equalsIgnoreCase("on")) {
+ return REPLICATED | LEGACY_TRUE;
+ }
+ if (v.equalsIgnoreCase("local")) {
+ return LOCAL;
+ }
+ if (v.equalsIgnoreCase("replicated")) {
+ return REPLICATED;
+ }
+ if (v.equalsIgnoreCase("local,replicated") || v.equalsIgnoreCase("replicated,local")) {
+ return LOCAL | REPLICATED;
+ }
+ return -1;
+ }
+
+ /**
+ * The X-QWP-Request-Durable-Ack header value for a tier set, or null when
+ * no tier is requested. A legacy set sends the shipped literal
+ * {@code "true"} so old servers keep recognizing it.
+ */
+ public static String requestHeaderValue(int tiers) {
+ if ((tiers & LEGACY_TRUE) != 0) {
+ return "true";
+ }
+ return explicitToken(tiers);
+ }
+
+ private static String explicitToken(int tiers) {
+ switch (tiers & (LOCAL | REPLICATED)) {
+ case LOCAL:
+ return "local";
+ case REPLICATED:
+ return "replicated";
+ case LOCAL | REPLICATED:
+ return "local,replicated";
+ default:
+ return null;
+ }
+ }
+}
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 41cc0a8c9..f3e40e410 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
@@ -384,7 +384,7 @@ public class QwpWebSocketSender implements Sender {
// values; Sender.build can override via the new connect overload.
private long reconnectMaxDurationMillis =
CursorWebSocketSendLoop.DEFAULT_RECONNECT_MAX_DURATION_MILLIS;
- private boolean requestDurableAck;
+ private int durableAckTiers;
// Monotonic per-attempt counter snapshotted onto every connection event
// fired from buildAndConnect. Counts every FOREGROUND endpoint try --
// successes and failures alike -- across this sender's lifetime.
@@ -493,7 +493,7 @@ public static QwpWebSocketSender connect(String host, int port, ClientTlsConfigu
host, port, tlsConfig,
DEFAULT_AUTO_FLUSH_ROWS, DEFAULT_AUTO_FLUSH_BYTES, DEFAULT_AUTO_FLUSH_INTERVAL_NANOS,
null,
- false, engine
+ DurableAckTiers.NONE, engine
);
} catch (Throwable t) {
try {
@@ -518,12 +518,12 @@ public static QwpWebSocketSender connect(
int autoFlushBytes,
long autoFlushIntervalNanos,
String authorizationHeader,
- boolean requestDurableAck,
+ int durableAckTiers,
CursorSendEngine cursorEngine
) {
return connect(host, port, tlsConfig, autoFlushRows, autoFlushBytes, autoFlushIntervalNanos,
authorizationHeader,
- requestDurableAck, cursorEngine, 5_000L);
+ durableAckTiers, cursorEngine, 5_000L);
}
/**
@@ -540,13 +540,13 @@ public static QwpWebSocketSender connect(
int autoFlushBytes,
long autoFlushIntervalNanos,
String authorizationHeader,
- boolean requestDurableAck,
+ int durableAckTiers,
CursorSendEngine cursorEngine,
long closeFlushTimeoutMillis
) {
return connect(host, port, tlsConfig, autoFlushRows, autoFlushBytes,
autoFlushIntervalNanos, authorizationHeader,
- requestDurableAck, cursorEngine,
+ durableAckTiers, cursorEngine,
closeFlushTimeoutMillis,
CursorWebSocketSendLoop.DEFAULT_RECONNECT_MAX_DURATION_MILLIS,
CursorWebSocketSendLoop.DEFAULT_RECONNECT_INITIAL_BACKOFF_MILLIS,
@@ -567,7 +567,7 @@ public static QwpWebSocketSender connect(
int autoFlushBytes,
long autoFlushIntervalNanos,
String authorizationHeader,
- boolean requestDurableAck,
+ int durableAckTiers,
CursorSendEngine cursorEngine,
long closeFlushTimeoutMillis,
long reconnectMaxDurationMillis,
@@ -576,7 +576,7 @@ public static QwpWebSocketSender connect(
) {
return connect(host, port, tlsConfig, autoFlushRows, autoFlushBytes,
autoFlushIntervalNanos, authorizationHeader,
- requestDurableAck, cursorEngine,
+ durableAckTiers, cursorEngine,
closeFlushTimeoutMillis, reconnectMaxDurationMillis,
reconnectInitialBackoffMillis, reconnectMaxBackoffMillis,
Sender.InitialConnectMode.OFF);
@@ -597,7 +597,7 @@ public static QwpWebSocketSender connect(
int autoFlushBytes,
long autoFlushIntervalNanos,
String authorizationHeader,
- boolean requestDurableAck,
+ int durableAckTiers,
CursorSendEngine cursorEngine,
long closeFlushTimeoutMillis,
long reconnectMaxDurationMillis,
@@ -607,7 +607,7 @@ public static QwpWebSocketSender connect(
) {
return connect(host, port, tlsConfig, autoFlushRows, autoFlushBytes,
autoFlushIntervalNanos, authorizationHeader,
- requestDurableAck, cursorEngine,
+ durableAckTiers, cursorEngine,
closeFlushTimeoutMillis, reconnectMaxDurationMillis,
reconnectInitialBackoffMillis, reconnectMaxBackoffMillis,
initialConnectMode, null, SenderErrorDispatcher.DEFAULT_CAPACITY);
@@ -626,7 +626,7 @@ public static QwpWebSocketSender connect(
int autoFlushBytes,
long autoFlushIntervalNanos,
String authorizationHeader,
- boolean requestDurableAck,
+ int durableAckTiers,
CursorSendEngine cursorEngine,
long closeFlushTimeoutMillis,
long reconnectMaxDurationMillis,
@@ -638,7 +638,7 @@ public static QwpWebSocketSender connect(
) {
return connect(host, port, tlsConfig, autoFlushRows, autoFlushBytes,
autoFlushIntervalNanos, authorizationHeader,
- requestDurableAck, cursorEngine,
+ durableAckTiers, cursorEngine,
closeFlushTimeoutMillis, reconnectMaxDurationMillis,
reconnectInitialBackoffMillis, reconnectMaxBackoffMillis,
initialConnectMode, errorHandler, errorInboxCapacity,
@@ -659,7 +659,7 @@ public static QwpWebSocketSender connect(
int autoFlushBytes,
long autoFlushIntervalNanos,
String authorizationHeader,
- boolean requestDurableAck,
+ int durableAckTiers,
CursorSendEngine cursorEngine,
long closeFlushTimeoutMillis,
long reconnectMaxDurationMillis,
@@ -674,7 +674,7 @@ public static QwpWebSocketSender connect(
singleEndpoint(host, port), tlsConfig,
autoFlushRows, autoFlushBytes, autoFlushIntervalNanos,
authorizationHeader,
- requestDurableAck, cursorEngine,
+ durableAckTiers, cursorEngine,
closeFlushTimeoutMillis, reconnectMaxDurationMillis,
reconnectInitialBackoffMillis, reconnectMaxBackoffMillis,
initialConnectMode, errorHandler, errorInboxCapacity,
@@ -697,7 +697,7 @@ public static QwpWebSocketSender connect(
int autoFlushBytes,
long autoFlushIntervalNanos,
String authorizationHeader,
- boolean requestDurableAck,
+ int durableAckTiers,
CursorSendEngine cursorEngine,
long closeFlushTimeoutMillis,
long reconnectMaxDurationMillis,
@@ -711,7 +711,7 @@ public static QwpWebSocketSender connect(
) {
return connectWithCredentialSupplier(endpoints, tlsConfig, autoFlushRows, autoFlushBytes,
autoFlushIntervalNanos, fixedAuthHeader(authorizationHeader),
- requestDurableAck, cursorEngine,
+ durableAckTiers, cursorEngine,
closeFlushTimeoutMillis, reconnectMaxDurationMillis,
reconnectInitialBackoffMillis, reconnectMaxBackoffMillis,
initialConnectMode, errorHandler, errorInboxCapacity,
@@ -737,7 +737,7 @@ public static QwpWebSocketSender connect(
int autoFlushBytes,
long autoFlushIntervalNanos,
String authorizationHeader,
- boolean requestDurableAck,
+ int durableAckTiers,
CursorSendEngine cursorEngine,
long closeFlushTimeoutMillis,
long reconnectMaxDurationMillis,
@@ -754,7 +754,7 @@ public static QwpWebSocketSender connect(
) {
return connectWithCredentialSupplier(endpoints, tlsConfig, autoFlushRows, autoFlushBytes,
autoFlushIntervalNanos, fixedAuthHeader(authorizationHeader),
- requestDurableAck, cursorEngine,
+ durableAckTiers, cursorEngine,
closeFlushTimeoutMillis, reconnectMaxDurationMillis,
reconnectInitialBackoffMillis, reconnectMaxBackoffMillis,
initialConnectMode, errorHandler, errorInboxCapacity,
@@ -780,7 +780,7 @@ public static QwpWebSocketSender connectWithCredentialSupplier(
int autoFlushBytes,
long autoFlushIntervalNanos,
Supplier
- * STATUS_DURABLE_ACK response format:
+ * STATUS_DURABLE_ACK / STATUS_LOCAL_DURABLE_ACK response format:
*
* This setting is only supported for WebSocket transport.
*
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 b34eb76a2..81859c475 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
@@ -816,9 +816,9 @@ private static boolean extractDurableAckConfirmed(String response, String expect
}
String value = response.substring(valueStart, lineEnd).trim();
// The server echoes the granted set verbatim (or the
- // historical "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.
+ // "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);
}
}
diff --git a/core/src/main/java/io/questdb/client/cutlass/qwp/client/DurableAckTiers.java b/core/src/main/java/io/questdb/client/cutlass/qwp/client/DurableAckTiers.java
index e6259914d..b3648cbec 100644
--- a/core/src/main/java/io/questdb/client/cutlass/qwp/client/DurableAckTiers.java
+++ b/core/src/main/java/io/questdb/client/cutlass/qwp/client/DurableAckTiers.java
@@ -33,10 +33,11 @@
*
* The server grants the full requested set or denies the request entirely
* (no confirmation header); it never substitutes a weaker guarantee. The
@@ -48,8 +49,8 @@ public final class DurableAckTiers {
public static final int NONE = 0;
public static final int LOCAL = 1;
public static final int REPLICATED = 2;
- // Modifier bit, only ever combined with REPLICATED: send the legacy "true"
- // request token and expect the historical "enabled" confirmation.
+ // Modifier bit, only ever combined with REPLICATED: send the "true"
+ // request token and expect the "enabled" confirmation.
public static final int LEGACY_TRUE = 4;
private DurableAckTiers() {
@@ -78,8 +79,8 @@ public static String configValue(int tiers) {
/**
* The X-QWP-Durable-Ack confirmation token the server must echo for this
* request, or null when no tier is requested. A legacy request expects
- * the historical {@code "enabled"} token; explicit requests expect their
- * own token set back verbatim.
+ * the {@code "enabled"} token; explicit tier requests expect their own
+ * token set back verbatim.
*/
public static String expectedConfirmToken(int tiers) {
if ((tiers & LEGACY_TRUE) != 0) {
@@ -108,8 +109,8 @@ public static boolean isTrimOnLocalAck(int tiers) {
/**
* Parses a {@code request_durable_ack} value into a tier set, or -1 for
- * an unrecognized value. {@code on} is the legacy alias for the
- * replicated tier (its shipped meaning); {@code off} is {@link #NONE}.
+ * an unrecognized value. {@code on} maps to the replicated tier with
+ * {@link #LEGACY_TRUE} set; {@code off} is {@link #NONE}.
*/
public static int parseConfigValue(CharSequence value) {
if (value == null) {
@@ -136,8 +137,8 @@ public static int parseConfigValue(CharSequence value) {
/**
* The X-QWP-Request-Durable-Ack header value for a tier set, or null when
- * no tier is requested. A legacy set sends the shipped literal
- * {@code "true"} so old servers keep recognizing it.
+ * no tier is requested. A legacy set sends the literal {@code "true"},
+ * the only request value servers without tier support recognize.
*/
public static String requestHeaderValue(int tiers) {
if ((tiers & LEGACY_TRUE) != 0) {
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 3e5a78d32..c94e94885 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
@@ -856,11 +856,12 @@ autoFlushIntervalNanos, fixedAuthHeader(authorizationHeader),
}
// ------------------------------------------------------------------
- // Binary-compatibility overloads. The durable-ack parameter was retyped
- // from boolean to the DurableAckTiers bitmask; callers compiled against
- // earlier releases still link against these boolean variants, which map
- // true to the legacy "true" request (the replicated tier, confirmed by
- // the historical "enabled" token) -- the shipped meaning of the opt-in.
+ // Boolean durable-ack overloads. Each maps its boolean flag onto the
+ // DurableAckTiers bitmask the master entry point takes: true becomes
+ // the legacy request (REPLICATED | LEGACY_TRUE -- header value "true",
+ // confirmed by the "enabled" token), false becomes NONE. They keep
+ // binary compatibility for callers linked against the boolean
+ // signatures.
// ------------------------------------------------------------------
public static QwpWebSocketSender connect(
diff --git a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/DurableAckTiersTest.java b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/DurableAckTiersTest.java
new file mode 100644
index 000000000..87b655653
--- /dev/null
+++ b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/DurableAckTiersTest.java
@@ -0,0 +1,148 @@
+/*******************************************************************************
+ * ___ _ ____ ____
+ * / _ \ _ _ ___ ___| |_| _ \| __ )
+ * | | | | | | |/ _ \/ __| __| | | | _ \
+ * | |_| | |_| | __/\__ \ |_| |_| | |_) |
+ * \__\_\\__,_|\___||___/\__|____/|____/
+ *
+ * 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.cutlass.qwp.client.DurableAckTiers;
+import org.junit.Test;
+
+import static io.questdb.client.cutlass.qwp.client.DurableAckTiers.LEGACY_TRUE;
+import static io.questdb.client.cutlass.qwp.client.DurableAckTiers.LOCAL;
+import static io.questdb.client.cutlass.qwp.client.DurableAckTiers.NONE;
+import static io.questdb.client.cutlass.qwp.client.DurableAckTiers.REPLICATED;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+
+/**
+ * Unit tests for the {@link DurableAckTiers} bitmask helpers: config-value
+ * parsing and printing, the request/confirmation wire tokens, and the
+ * trim-trigger predicate.
+ */
+public class DurableAckTiersTest {
+
+ @Test
+ public void testConfigValueInverseOfParse() {
+ // configValue . parseConfigValue is identity for every accepted value
+ // (modulo case and the replicated,local ordering alias).
+ String[] canonical = {"off", "on", "local", "replicated", "local,replicated"};
+ for (String v : canonical) {
+ assertEquals(v, DurableAckTiers.configValue(DurableAckTiers.parseConfigValue(v)));
+ }
+ // The reversed alias normalizes to the canonical order.
+ assertEquals("local,replicated",
+ DurableAckTiers.configValue(DurableAckTiers.parseConfigValue("replicated,local")));
+ }
+
+ @Test
+ public void testConfigValuePrintsTierSets() {
+ assertEquals("off", DurableAckTiers.configValue(NONE));
+ assertEquals("local", DurableAckTiers.configValue(LOCAL));
+ assertEquals("replicated", DurableAckTiers.configValue(REPLICATED));
+ assertEquals("local,replicated", DurableAckTiers.configValue(LOCAL | REPLICATED));
+ assertEquals("on", DurableAckTiers.configValue(REPLICATED | LEGACY_TRUE));
+ }
+
+ @Test
+ public void testExpectedConfirmToken() {
+ // Explicit requests are confirmed by their own token echoed back;
+ // the legacy request is confirmed by "enabled"; no request expects
+ // no confirmation.
+ assertNull(DurableAckTiers.expectedConfirmToken(NONE));
+ assertEquals("local", DurableAckTiers.expectedConfirmToken(LOCAL));
+ assertEquals("replicated", DurableAckTiers.expectedConfirmToken(REPLICATED));
+ assertEquals("local,replicated", DurableAckTiers.expectedConfirmToken(LOCAL | REPLICATED));
+ assertEquals("enabled", DurableAckTiers.expectedConfirmToken(REPLICATED | LEGACY_TRUE));
+ }
+
+ @Test
+ public void testHasLocalHasReplicated() {
+ assertFalse(DurableAckTiers.hasLocal(NONE));
+ assertFalse(DurableAckTiers.hasReplicated(NONE));
+ assertTrue(DurableAckTiers.hasLocal(LOCAL));
+ assertFalse(DurableAckTiers.hasReplicated(LOCAL));
+ assertFalse(DurableAckTiers.hasLocal(REPLICATED));
+ assertTrue(DurableAckTiers.hasReplicated(REPLICATED));
+ assertTrue(DurableAckTiers.hasLocal(LOCAL | REPLICATED));
+ assertTrue(DurableAckTiers.hasReplicated(LOCAL | REPLICATED));
+ assertTrue(DurableAckTiers.hasReplicated(REPLICATED | LEGACY_TRUE));
+ }
+
+ @Test
+ public void testIsTrimOnLocalAck() {
+ // Only a local-without-replicated request trims on the local ack;
+ // any set that includes the replicated tier trims on the replicated
+ // ack (strongest requested guarantee wins).
+ assertTrue(DurableAckTiers.isTrimOnLocalAck(LOCAL));
+ assertFalse(DurableAckTiers.isTrimOnLocalAck(NONE));
+ assertFalse(DurableAckTiers.isTrimOnLocalAck(REPLICATED));
+ assertFalse(DurableAckTiers.isTrimOnLocalAck(LOCAL | REPLICATED));
+ assertFalse(DurableAckTiers.isTrimOnLocalAck(REPLICATED | LEGACY_TRUE));
+ }
+
+ @Test
+ public void testParseConfigValueAccepted() {
+ assertEquals(NONE, DurableAckTiers.parseConfigValue("off"));
+ assertEquals(REPLICATED | LEGACY_TRUE, DurableAckTiers.parseConfigValue("on"));
+ assertEquals(LOCAL, DurableAckTiers.parseConfigValue("local"));
+ assertEquals(REPLICATED, DurableAckTiers.parseConfigValue("replicated"));
+ assertEquals(LOCAL | REPLICATED, DurableAckTiers.parseConfigValue("local,replicated"));
+ assertEquals(LOCAL | REPLICATED, DurableAckTiers.parseConfigValue("replicated,local"));
+ }
+
+ @Test
+ public void testParseConfigValueCaseInsensitiveAndTrimmed() {
+ assertEquals(REPLICATED | LEGACY_TRUE, DurableAckTiers.parseConfigValue("ON"));
+ assertEquals(LOCAL, DurableAckTiers.parseConfigValue("Local"));
+ assertEquals(LOCAL | REPLICATED, DurableAckTiers.parseConfigValue("LOCAL,REPLICATED"));
+ assertEquals(NONE, DurableAckTiers.parseConfigValue(" off "));
+ }
+
+ @Test
+ public void testParseConfigValueRejected() {
+ // The whole value must match one of the accepted spellings; spaces
+ // inside a list, empty and unknown tokens, and null all read as -1.
+ assertEquals(-1, DurableAckTiers.parseConfigValue(null));
+ assertEquals(-1, DurableAckTiers.parseConfigValue(""));
+ assertEquals(-1, DurableAckTiers.parseConfigValue("true"));
+ assertEquals(-1, DurableAckTiers.parseConfigValue("yes"));
+ assertEquals(-1, DurableAckTiers.parseConfigValue("enabled"));
+ assertEquals(-1, DurableAckTiers.parseConfigValue("local, replicated"));
+ assertEquals(-1, DurableAckTiers.parseConfigValue("local,"));
+ assertEquals(-1, DurableAckTiers.parseConfigValue("local,local"));
+ assertEquals(-1, DurableAckTiers.parseConfigValue("remote"));
+ }
+
+ @Test
+ public void testRequestHeaderValue() {
+ // The legacy set is sent as the literal "true"; explicit sets send
+ // their token list; no request means no header.
+ assertNull(DurableAckTiers.requestHeaderValue(NONE));
+ assertEquals("local", DurableAckTiers.requestHeaderValue(LOCAL));
+ assertEquals("replicated", DurableAckTiers.requestHeaderValue(REPLICATED));
+ assertEquals("local,replicated", DurableAckTiers.requestHeaderValue(LOCAL | REPLICATED));
+ assertEquals("true", DurableAckTiers.requestHeaderValue(REPLICATED | LEGACY_TRUE));
+ }
+}
diff --git a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/WebSocketResponseLocalDurableAckTest.java b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/WebSocketResponseLocalDurableAckTest.java
new file mode 100644
index 000000000..419a82392
--- /dev/null
+++ b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/WebSocketResponseLocalDurableAckTest.java
@@ -0,0 +1,109 @@
+/*******************************************************************************
+ * ___ _ ____ ____
+ * / _ \ _ _ ___ ___| |_| _ \| __ )
+ * | | | | | | |/ _ \/ __| __| | | | _ \
+ * | |_| | |_| | __/\__ \ |_| |_| | |_) |
+ * \__\_\\__,_|\___||___/\__|____/|____/
+ *
+ * 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.cutlass.qwp.client.WebSocketResponse;
+import io.questdb.client.std.MemoryTag;
+import io.questdb.client.std.Unsafe;
+import io.questdb.client.test.tools.TestUtils;
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+/**
+ * Wire-format tests for the {@code STATUS_LOCAL_DURABLE_ACK} response: it
+ * shares the sequence-less {@code status + tableCount + entries} layout with
+ * {@code STATUS_DURABLE_ACK} and must survive a write/read round trip,
+ * validate structurally, and classify via {@code isLocalDurableAck()} only.
+ */
+public class WebSocketResponseLocalDurableAckTest {
+
+ @Test
+ public void testClassificationIsMutuallyExclusive() {
+ WebSocketResponse local = WebSocketResponse.localDurableAck("trades", 7L);
+ assertTrue(local.isLocalDurableAck());
+ assertFalse(local.isDurableAck());
+ assertFalse(local.isSuccess());
+ assertEquals("LOCAL_DURABLE_ACK", local.getStatusName());
+
+ WebSocketResponse replicated = WebSocketResponse.durableAck("trades", 7L);
+ assertTrue(replicated.isDurableAck());
+ assertFalse(replicated.isLocalDurableAck());
+ assertEquals("DURABLE_ACK", replicated.getStatusName());
+ }
+
+ @Test
+ public void testStructurallyValid() throws Exception {
+ TestUtils.assertMemoryLeak(() -> {
+ WebSocketResponse response = WebSocketResponse.localDurableAck("trades", 42L);
+ int size = response.serializedSize();
+ long ptr = Unsafe.malloc(size, MemoryTag.NATIVE_DEFAULT);
+ try {
+ assertEquals(size, response.writeTo(ptr));
+ assertTrue(WebSocketResponse.isStructurallyValid(ptr, size));
+ // A truncated frame (entry cut short) must not validate.
+ assertFalse(WebSocketResponse.isStructurallyValid(ptr, size - 1));
+ } finally {
+ Unsafe.free(ptr, size, MemoryTag.NATIVE_DEFAULT);
+ }
+ });
+ }
+
+ @Test
+ public void testToStringNamesLocalStatus() {
+ WebSocketResponse response = WebSocketResponse.localDurableAck("trades", 1L);
+ assertEquals("WebSocketResponse{status=LOCAL_DURABLE_ACK, tables=1}", response.toString());
+ }
+
+ @Test
+ public void testWriteReadRoundTrip() throws Exception {
+ TestUtils.assertMemoryLeak(() -> {
+ WebSocketResponse out = WebSocketResponse.localDurableAck("trades", 42L);
+ int size = out.serializedSize();
+ long ptr = Unsafe.malloc(size, MemoryTag.NATIVE_DEFAULT);
+ try {
+ assertEquals(size, out.writeTo(ptr));
+
+ WebSocketResponse in = new WebSocketResponse();
+ assertTrue(in.readFrom(ptr, size));
+ assertEquals(WebSocketResponse.STATUS_LOCAL_DURABLE_ACK, in.getStatus());
+ assertTrue(in.isLocalDurableAck());
+ assertFalse("local ack must not classify as the replicated ack",
+ in.isDurableAck());
+ assertEquals(-1L, in.getSequence());
+ assertEquals(1, in.getTableEntryCount());
+ assertEquals("trades", in.getTableName(0));
+ assertEquals(42L, in.getTableSeqTxn(0));
+
+ // Truncated input must be rejected, not misparsed.
+ assertFalse(new WebSocketResponse().readFrom(ptr, size - 1));
+ } finally {
+ Unsafe.free(ptr, size, MemoryTag.NATIVE_DEFAULT);
+ }
+ });
+ }
+}
diff --git a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/DurableAckIntegrationTest.java b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/DurableAckIntegrationTest.java
index 3378f7177..a50e5ffc5 100644
--- a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/DurableAckIntegrationTest.java
+++ b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/DurableAckIntegrationTest.java
@@ -27,6 +27,10 @@
import io.questdb.client.Sender;
import io.questdb.client.cutlass.line.LineSenderException;
import io.questdb.client.cutlass.qwp.client.QwpDurableAckMismatchException;
+import io.questdb.client.cutlass.qwp.client.QwpWebSocketSender;
+import io.questdb.client.cutlass.qwp.client.WebSocketResponse;
+import io.questdb.client.cutlass.qwp.client.sf.cursor.CursorSendEngine;
+import io.questdb.client.cutlass.qwp.client.sf.cursor.CursorWebSocketSendLoop;
import io.questdb.client.std.Files;
import io.questdb.client.test.cutlass.qwp.websocket.TestWebSocketServer;
import io.questdb.client.test.tools.TestUtils;
@@ -125,6 +129,224 @@ public void testConnectStringOnRequiresServerSupport() throws Exception {
});
}
+ @Test
+ public void testBooleanConnectOverloadRequestsLegacyTier() throws Exception {
+ // The boolean connect overloads map true onto the legacy request:
+ // the upgrade header carries "true" and the "enabled" grant is
+ // accepted. Callers linked against the boolean signatures keep the
+ // pre-tier wire behavior. (ExportedApiCompatibilityTest pins the
+ // signatures; this pins what they do.)
+ TestUtils.assertMemoryLeak(() -> {
+ Assert.assertEquals(0, Files.mkdir(sfDir, Files.DIR_MODE_DEFAULT));
+ DurableAckCapableHandler handler = new DurableAckCapableHandler();
+ try (TestWebSocketServer server = new TestWebSocketServer(handler, true)) {
+ server.start();
+ Assert.assertTrue(server.awaitStart(5, TimeUnit.SECONDS));
+ int port = server.getPort();
+
+ CursorSendEngine engine = new CursorSendEngine(sfDir, 16384);
+ try (Sender ignored = QwpWebSocketSender.connect(
+ "localhost", port, null, 0, 0, 0L, null, true, engine)) {
+ Assert.assertEquals("true",
+ server.pollDurableAckRequest(5, TimeUnit.SECONDS));
+ }
+ }
+ });
+ }
+
+ @Test
+ public void testBuilderInvalidTierValueRejected() {
+ // The programmatic CharSequence overload applies the same parse as the
+ // config string: a typo must throw, naming the key and the value.
+ try {
+ Sender.builder(Sender.Transport.WEBSOCKET).requestDurableAck("yes");
+ Assert.fail("expected LineSenderException for invalid tier value");
+ } catch (LineSenderException e) {
+ Assert.assertTrue(
+ "message names the offending key+value, was: " + e.getMessage(),
+ e.getMessage().contains("request_durable_ack")
+ && e.getMessage().contains("yes"));
+ }
+ }
+
+ @Test
+ public void testBuilderRejectsTiersOnHttpTransport() {
+ // Durable-ack streams exist only on the WebSocket transport; asking
+ // for a tier on an HTTP builder must fail fast at configuration time.
+ try {
+ Sender.builder(Sender.Transport.HTTP).requestDurableAck("local");
+ Assert.fail("expected LineSenderException for HTTP transport");
+ } catch (LineSenderException e) {
+ Assert.assertTrue("was: " + e.getMessage(),
+ e.getMessage().contains("only supported for WebSocket"));
+ }
+ }
+
+ @Test
+ public void testConnectStringLocalRequiresServerSupport() throws Exception {
+ // Tier requests keep the all-or-nothing contract of the legacy opt-in:
+ // a server that does not confirm the durable-ack grant (no
+ // X-QWP-Durable-Ack header) must fail the connect, not leave the
+ // store-and-forward log growing while waiting on acks that never come.
+ TestUtils.assertMemoryLeak(() -> {
+ DurableAckCapableHandler handler = new DurableAckCapableHandler();
+ try (TestWebSocketServer server = new TestWebSocketServer(handler, false)) {
+ server.start();
+ Assert.assertTrue(server.awaitStart(5, TimeUnit.SECONDS));
+
+ int port = server.getPort();
+ String config = "ws::addr=localhost:" + port + ";sf_dir=" + sfDir + ";request_durable_ack=local;";
+ try (Sender ignored = Sender.fromConfig(config)) {
+ Assert.fail("expected connect to fail with QwpDurableAckMismatchException");
+ } catch (QwpDurableAckMismatchException e) {
+ Assert.assertEquals("localhost", e.getHost());
+ Assert.assertEquals(port, e.getPort());
+ }
+ }
+ });
+ }
+
+ @Test
+ public void testLocalRequestDeniedWhenServerGrantsDifferentSet() throws Exception {
+ // A server that answers a "local" request with the legacy "enabled"
+ // token granted a set other than the one requested. The client must
+ // read any token except its expected one as a denial -- trimming on a
+ // foreign grant could otherwise drop data on a guarantee weaker than
+ // the caller configured.
+ TestUtils.assertMemoryLeak(() -> {
+ DurableAckCapableHandler handler = new DurableAckCapableHandler();
+ try (TestWebSocketServer server = new TestWebSocketServer(handler, true)) {
+ server.setDurableAckHeaderValue("enabled");
+ server.start();
+ Assert.assertTrue(server.awaitStart(5, TimeUnit.SECONDS));
+
+ int port = server.getPort();
+ String config = "ws::addr=localhost:" + port + ";sf_dir=" + sfDir + ";request_durable_ack=local;";
+ try (Sender ignored = Sender.fromConfig(config)) {
+ Assert.fail("expected connect to fail with QwpDurableAckMismatchException");
+ } catch (QwpDurableAckMismatchException e) {
+ Assert.assertEquals(port, e.getPort());
+ }
+ }
+ });
+ }
+
+ @Test
+ public void testRequestHeaderCarriesConfiguredTierSet() throws Exception {
+ // The upgrade request must carry the exact token for each configured
+ // set: the legacy "on" travels as "true" (the request value
+ // tier-unaware servers recognize), explicit sets travel verbatim.
+ // No rows are sent, so close() returns without waiting on acks.
+ TestUtils.assertMemoryLeak(() -> {
+ DurableAckCapableHandler handler = new DurableAckCapableHandler();
+ try (TestWebSocketServer server = new TestWebSocketServer(handler, true)) {
+ server.start();
+ Assert.assertTrue(server.awaitStart(5, TimeUnit.SECONDS));
+ int port = server.getPort();
+
+ String[][] cases = {
+ {"on", "true"},
+ {"local", "local"},
+ {"replicated", "replicated"},
+ {"local,replicated", "local,replicated"},
+ };
+ for (String[] c : cases) {
+ String config = "ws::addr=localhost:" + port + ";sf_dir=" + sfDir
+ + ";request_durable_ack=" + c[0] + ";";
+ Sender.fromConfig(config).close();
+ Assert.assertEquals("config value " + c[0],
+ c[1], server.pollDurableAckRequest(5, TimeUnit.SECONDS));
+ }
+ }
+ });
+ }
+
+ @Test
+ public void testEndToEndBothTiersLocalAckIsProgressOnly() throws Exception {
+ // request_durable_ack=local,replicated: the replicated ack is the trim
+ // trigger; local acks surface early per-table progress without popping
+ // anything. The loop's counters and the per-table local watermark are
+ // the observable surface for that split.
+ TestUtils.assertMemoryLeak(() -> {
+ DurableAckCapableHandler handler = new DurableAckCapableHandler();
+ try (TestWebSocketServer server = new TestWebSocketServer(handler, true)) {
+ server.start();
+ Assert.assertTrue(server.awaitStart(5, TimeUnit.SECONDS));
+
+ int port = server.getPort();
+ String config = "ws::addr=localhost:" + port + ";sf_dir=" + sfDir
+ + ";request_durable_ack=local,replicated;close_flush_timeout_millis=5000;";
+ try (Sender sender = Sender.fromConfig(config)) {
+ for (int i = 0; i < 10; i++) {
+ sender.table("trades").longColumn("v", i).atNow();
+ }
+ sender.flush(); // one batch -> one OK
+ handler.awaitOkBatches(1);
+ long batches = 1;
+
+ CursorWebSocketSendLoop loop =
+ ((QwpWebSocketSender) sender).cursorSendLoopForTest();
+ Assert.assertNotNull(loop);
+
+ // Release local acks covering everything OK'd, then nudge
+ // the connection with extra rows until the I/O thread has
+ // observed one -- the local ack alone must not trim.
+ handler.emitLocalDurableAckForAll();
+ long deadline = System.currentTimeMillis() + 5000;
+ while (loop.getTotalLocalDurableAcks() == 0
+ && System.currentTimeMillis() < deadline) {
+ sender.table("trades").longColumn("v", -1L).atNow();
+ sender.flush();
+ batches++;
+ Thread.sleep(10);
+ }
+ Assert.assertTrue("local ack never observed",
+ loop.getTotalLocalDurableAcks() > 0);
+ Assert.assertTrue("local watermark tracks the fsync frontier",
+ loop.getLocalDurableTableWatermark("trades") >= 0);
+ Assert.assertEquals(
+ "local acks must not advance trim when replicated is requested",
+ 0L, loop.getTotalDurableTrimAdvances());
+
+ // Cover every batch sent (nudges included) with a
+ // replicated ack so close() drains on the actual trim
+ // trigger.
+ handler.awaitOkBatches(batches);
+ handler.emitDurableAckForAll();
+ }
+ }
+ });
+ }
+
+ @Test
+ public void testEndToEndLocalTrimDefersUntilLocalAck() throws Exception {
+ // request_durable_ack=local: OK frames alone never trim; the
+ // STATUS_LOCAL_DURABLE_ACK stream is the trim trigger. close() drains
+ // only once the local ack covers everything sent -- the local-tier
+ // mirror of testEndToEndDurableTrimDefersUntilUploadAck.
+ TestUtils.assertMemoryLeak(() -> {
+ DurableAckCapableHandler handler = new DurableAckCapableHandler();
+ try (TestWebSocketServer server = new TestWebSocketServer(handler, true)) {
+ server.start();
+ Assert.assertTrue(server.awaitStart(5, TimeUnit.SECONDS));
+
+ int port = server.getPort();
+ String config = "ws::addr=localhost:" + port + ";sf_dir=" + sfDir
+ + ";request_durable_ack=local;close_flush_timeout_millis=5000;";
+ try (Sender sender = Sender.fromConfig(config)) {
+ for (int i = 0; i < 50; i++) {
+ sender.table("trades").longColumn("v", i).atNow();
+ }
+ sender.flush(); // one batch -> one OK
+ handler.awaitOkBatches(1);
+ handler.emitLocalDurableAckForAll();
+ }
+ // close() returned without timing out: the local ack drove the
+ // trim to completion.
+ }
+ });
+ }
+
@Test
public void testEndToEndDurableTrimDefersUntilUploadAck() throws Exception {
// Server confirms support and emits OK acks but no durable-acks at first.
@@ -147,12 +369,12 @@ public void testEndToEndDurableTrimDefersUntilUploadAck() throws Exception {
}
sender.flush();
- // Wait for the server to OK every batch so we know the OK
- // watermark is fully advanced. Without a durable-ack the
- // client's ackedFsn must still be behind publishedFsn --
+ // Wait for the server to OK the flushed batch so we know
+ // the OK watermark is fully advanced. Without a durable-ack
+ // the client's ackedFsn must still be behind publishedFsn --
// we don't assert on internals here, just observe that
// the contract holds at the boundary check below.
- handler.awaitOks();
+ handler.awaitOkBatches(1);
// Release a cumulative durable-ack covering everything that
// has been OK'd so far. The client's I/O thread reads new
@@ -169,10 +391,12 @@ public void testEndToEndDurableTrimDefersUntilUploadAck() throws Exception {
});
}
- private static byte[] buildDurableAckFrame(long seqTxn) {
+ private static byte[] buildDurableAckFrame(byte status, long seqTxn) {
+ // STATUS_DURABLE_ACK and STATUS_LOCAL_DURABLE_ACK share the layout:
+ // status(1) + tableCount(2) + nameLen(2) + name + seqTxn(8).
byte[] name = DurableAckCapableHandler.TABLE_NAME.getBytes(StandardCharsets.UTF_8);
ByteBuffer bb = ByteBuffer.allocate(1 + 2 + 2 + name.length + 8).order(ByteOrder.LITTLE_ENDIAN);
- bb.put((byte) 0x02); // STATUS_DURABLE_ACK
+ bb.put(status);
bb.putShort((short) 1); // tableCount
bb.putShort((short) name.length);
bb.put(name);
@@ -230,20 +454,36 @@ public void onBinaryMessage(TestWebSocketServer.ClientHandler client, byte[] dat
}
}
- void awaitOks() throws InterruptedException {
+ void awaitOkBatches(long count) throws InterruptedException {
+ // One OK frame per QWP batch (i.e. per flush), NOT per row. A
+ // silent return on timeout would let the caller proceed on a
+ // watermark that never advanced, so this fails loudly instead.
long deadline = System.currentTimeMillis() + (long) 5000;
- while (totalOks() < (long) 50 && System.currentTimeMillis() < deadline) {
+ while (totalOks() < count && System.currentTimeMillis() < deadline) {
Thread.sleep(10);
}
+ Assert.assertTrue(
+ "server never OK'd " + count + " batch(es), got " + totalOks(),
+ totalOks() >= count);
}
void emitDurableAckForAll() throws IOException {
// Cumulative durable-ack: every OK already issued is now durable.
// Single-table handler so one entry suffices.
+ emitAckForAll(WebSocketResponse.STATUS_DURABLE_ACK);
+ }
+
+ void emitLocalDurableAckForAll() throws IOException {
+ // Cumulative local-durability ack: every OK already issued is now
+ // fdatasync-durable on the "server".
+ emitAckForAll(WebSocketResponse.STATUS_LOCAL_DURABLE_ACK);
+ }
+
+ private void emitAckForAll(byte status) throws IOException {
TestWebSocketServer.ClientHandler c = activeClient;
if (c != null) {
long seqTxn = Math.max(0L, nextSeqTxn.get() - 1L);
- c.sendBinary(buildDurableAckFrame(seqTxn));
+ c.sendBinary(buildDurableAckFrame(status, seqTxn));
}
}
diff --git a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoopDurableAckTest.java b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoopDurableAckTest.java
index b52ed76a8..dcf4eb40f 100644
--- a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoopDurableAckTest.java
+++ b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoopDurableAckTest.java
@@ -292,6 +292,146 @@ public void testDurableModePartialCoverageDoesNotAdvance() throws Exception {
});
}
+ @Test
+ public void testBothTiersLocalAckIsProgressOnly() throws Exception {
+ // With local,replicated requested the trim trigger is the replicated
+ // ack. A local ack must not pop the pending queue or move ackedFsn;
+ // it only bumps the local counter and the per-table local watermark.
+ // The replicated ack then trims as usual.
+ TestUtils.assertMemoryLeak(() -> {
+ try (CursorSendEngine engine = newEngine()) {
+ appendFrames(engine, 1);
+ CursorWebSocketSendLoop loop = newBothTiersLoop(engine);
+ setSentCount(loop, 1);
+ deliverOk(loop, 0, names("trades"), txns(7L));
+
+ deliverLocalDurableAck(loop, names("trades"), txns(7L));
+ assertEquals("local ack must not trim when replicated is requested",
+ -1L, engine.ackedFsn());
+ assertEquals(1, pendingSize(loop));
+ assertEquals(1L, loop.getTotalLocalDurableAcks());
+ assertEquals(0L, loop.getTotalDurableTrimAdvances());
+ assertEquals("local watermark records the fsync frontier",
+ 7L, loop.getLocalDurableTableWatermark("trades"));
+
+ deliverDurableAck(loop, names("trades"), txns(7L));
+ assertEquals(0L, engine.ackedFsn());
+ assertEquals(0, pendingSize(loop));
+ assertEquals(1L, loop.getTotalDurableAcks());
+ assertEquals(1L, loop.getTotalDurableTrimAdvances());
+ }
+ });
+ }
+
+ @Test
+ public void testBothTiersLocalWatermarkIsMonotonic() throws Exception {
+ // A delayed/duplicate local ack naming a smaller seqTxn must not move
+ // the per-table local watermark backwards; the frame still counts.
+ TestUtils.assertMemoryLeak(() -> {
+ try (CursorSendEngine engine = newEngine()) {
+ CursorWebSocketSendLoop loop = newBothTiersLoop(engine);
+
+ deliverLocalDurableAck(loop, names("trades"), txns(10L));
+ assertEquals(10L, loop.getLocalDurableTableWatermark("trades"));
+
+ deliverLocalDurableAck(loop, names("trades"), txns(5L));
+ assertEquals("stale local ack must not unwind the watermark",
+ 10L, loop.getLocalDurableTableWatermark("trades"));
+ assertEquals(2L, loop.getTotalLocalDurableAcks());
+
+ assertEquals("unseen table reads as -1",
+ -1L, loop.getLocalDurableTableWatermark("orders"));
+ }
+ });
+ }
+
+ @Test
+ public void testDefaultModeIgnoresStrayLocalDurableAck() throws Exception {
+ // Without any opt-in a STATUS_LOCAL_DURABLE_ACK frame is logged and
+ // dropped: no trim, no counter movement.
+ TestUtils.assertMemoryLeak(() -> {
+ try (CursorSendEngine engine = newEngine()) {
+ appendFrames(engine, 1);
+ CursorWebSocketSendLoop loop = newDefaultLoop(engine);
+ setSentCount(loop, 1);
+ deliverLocalDurableAck(loop, names("anything"), txns(99L));
+ assertEquals(-1L, engine.ackedFsn());
+ assertEquals(0L, loop.getTotalLocalDurableAcks());
+ }
+ });
+ }
+
+ @Test
+ public void testLocalOnlyModeLocalAckAdvancesTrim() throws Exception {
+ // With only the local tier requested, STATUS_LOCAL_DURABLE_ACK is the
+ // trim trigger: an OK queues the entry and the local ack drains it
+ // through the same watermark path the replicated ack drives otherwise.
+ // The progress-only watermark map stays untouched in this mode.
+ TestUtils.assertMemoryLeak(() -> {
+ try (CursorSendEngine engine = newEngine()) {
+ appendFrames(engine, 1);
+ CursorWebSocketSendLoop loop = newLocalLoop(engine);
+ setSentCount(loop, 1);
+
+ deliverOk(loop, 0, names("trades"), txns(7L));
+ assertEquals("OK alone must not trim in local-only mode",
+ -1L, engine.ackedFsn());
+ assertEquals(1, pendingSize(loop));
+
+ deliverLocalDurableAck(loop, names("trades"), txns(7L));
+ assertEquals(0L, engine.ackedFsn());
+ assertEquals(0, pendingSize(loop));
+ assertEquals(1L, loop.getTotalLocalDurableAcks());
+ assertEquals(0L, loop.getTotalDurableAcks());
+ assertEquals(1L, loop.getTotalDurableTrimAdvances());
+ assertEquals("local-only mode feeds the trim watermarks, not the progress map",
+ -1L, loop.getLocalDurableTableWatermark("trades"));
+ }
+ });
+ }
+
+ @Test
+ public void testLocalOnlyModeReplicatedAckAlsoTrims() throws Exception {
+ // The server grants all-or-nothing, so a local-only connection should
+ // never see STATUS_DURABLE_ACK -- but if one arrives, trimming on it
+ // is safe: object-store durability subsumes the local-fsync guarantee
+ // the caller asked for.
+ TestUtils.assertMemoryLeak(() -> {
+ try (CursorSendEngine engine = newEngine()) {
+ appendFrames(engine, 1);
+ CursorWebSocketSendLoop loop = newLocalLoop(engine);
+ setSentCount(loop, 1);
+ deliverOk(loop, 0, names("trades"), txns(7L));
+
+ deliverDurableAck(loop, names("trades"), txns(7L));
+ assertEquals(0L, engine.ackedFsn());
+ assertEquals(1L, loop.getTotalDurableAcks());
+ assertEquals(0L, loop.getTotalLocalDurableAcks());
+ }
+ });
+ }
+
+ @Test
+ public void testReconnectClearsLocalWatermarks() throws Exception {
+ // clearDurableAckTracking (invoked on every client swap) must drop the
+ // local progress watermarks along with the trim state: the new
+ // connection's server rebuilds both frontiers from scratch.
+ TestUtils.assertMemoryLeak(() -> {
+ try (CursorSendEngine engine = newEngine()) {
+ CursorWebSocketSendLoop loop = newBothTiersLoop(engine);
+ deliverLocalDurableAck(loop, names("trades"), txns(42L));
+ assertEquals(42L, loop.getLocalDurableTableWatermark("trades"));
+
+ Method m = CursorWebSocketSendLoop.class.getDeclaredMethod("clearDurableAckTracking");
+ m.setAccessible(true);
+ m.invoke(loop);
+
+ assertEquals("stale local watermarks must not survive a reconnect",
+ -1L, loop.getLocalDurableTableWatermark("trades"));
+ }
+ });
+ }
+
@Test
public void testNackInDurableModeIsTerminalAndDoesNotAdvanceTrim() throws Exception {
// A SCHEMA_MISMATCH NACK is TERMINAL: it latches the typed error and
@@ -543,13 +683,14 @@ private static void appendFrames(CursorSendEngine engine, int count) {
}
}
- private static long buildDurableAckPayload(String[] tableNames, long[] seqTxns) {
- // STATUS_DURABLE_ACK frame: status(1) + tableCount(2) + entries(nameLen(2)+name+seqTxn(8))
+ private static long buildDurableAckPayload(byte status, String[] tableNames, long[] seqTxns) {
+ // Durable-ack frame (STATUS_DURABLE_ACK or STATUS_LOCAL_DURABLE_ACK,
+ // same layout): status(1) + tableCount(2) + entries(nameLen(2)+name+seqTxn(8))
int size = 3;
for (String t : tableNames) size += 2 + t.getBytes(StandardCharsets.UTF_8).length + 8;
long ptr = Unsafe.malloc(size, MemoryTag.NATIVE_DEFAULT);
int offset = 0;
- Unsafe.getUnsafe().putByte(ptr + offset, WebSocketResponse.STATUS_DURABLE_ACK);
+ Unsafe.getUnsafe().putByte(ptr + offset, status);
offset += 1;
Unsafe.getUnsafe().putShort(ptr + offset, (short) tableNames.length);
offset += 2;
@@ -610,7 +751,15 @@ private static long buildOkPayload(long wireSeq, String[] tableNames, long[] seq
}
private static void deliverDurableAck(CursorWebSocketSendLoop loop, String[] tableNames, long[] seqTxns) throws Exception {
- long packed = buildDurableAckPayload(tableNames, seqTxns);
+ deliverAck(loop, WebSocketResponse.STATUS_DURABLE_ACK, tableNames, seqTxns);
+ }
+
+ private static void deliverLocalDurableAck(CursorWebSocketSendLoop loop, String[] tableNames, long[] seqTxns) throws Exception {
+ deliverAck(loop, WebSocketResponse.STATUS_LOCAL_DURABLE_ACK, tableNames, seqTxns);
+ }
+
+ private static void deliverAck(CursorWebSocketSendLoop loop, byte status, String[] tableNames, long[] seqTxns) throws Exception {
+ long packed = buildDurableAckPayload(status, tableNames, seqTxns);
long ptr = packed & 0xFFFFFFFFFFFFL;
int size = (int) (packed >>> 48);
try {
@@ -664,21 +813,28 @@ private CursorSendEngine newEngine() {
}
private CursorWebSocketSendLoop newDefaultLoop(CursorSendEngine engine) {
- return new CursorWebSocketSendLoop(
- null, engine, 0L, CursorWebSocketSendLoop.DEFAULT_PARK_NANOS,
- () -> {
- throw new UnsupportedOperationException("test loop is never started");
- },
- 100L, 5_000L, DurableAckTiers.NONE);
+ return newLoop(engine, DurableAckTiers.NONE);
}
private CursorWebSocketSendLoop newDurableLoop(CursorSendEngine engine) {
+ return newLoop(engine, DurableAckTiers.REPLICATED);
+ }
+
+ private CursorWebSocketSendLoop newLocalLoop(CursorSendEngine engine) {
+ return newLoop(engine, DurableAckTiers.LOCAL);
+ }
+
+ private CursorWebSocketSendLoop newBothTiersLoop(CursorSendEngine engine) {
+ return newLoop(engine, DurableAckTiers.LOCAL | DurableAckTiers.REPLICATED);
+ }
+
+ private CursorWebSocketSendLoop newLoop(CursorSendEngine engine, int durableAckTiers) {
return new CursorWebSocketSendLoop(
null, engine, 0L, CursorWebSocketSendLoop.DEFAULT_PARK_NANOS,
() -> {
throw new UnsupportedOperationException("test loop is never started");
},
- 100L, 5_000L, DurableAckTiers.REPLICATED);
+ 100L, 5_000L, durableAckTiers);
}
private static int pendingSize(CursorWebSocketSendLoop loop) throws Exception {
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 908e3bd56..8e6e35b9a 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
@@ -60,6 +60,10 @@ public class TestWebSocketServer implements Closeable {
// Authorization header value captured from each well-formed upgrade request ("" when absent), in
// arrival order. Tests poll this to assert the token a provider supplied at each (re)handshake.
private final BlockingQueue
+ *
+ * {@link #LEGACY_TRUE} is a modifier bit recording that the request travels
+ * as the shipped literal {@code "true"} header value, whose grant the server
+ * confirms with the historical {@code "enabled"} token. It always combines
+ * with {@link #REPLICATED} — the shipped meaning of the legacy opt-in.
+ *
* +--------+------------+--------------------------------------+
* | status | tableCount | table entries |
@@ -80,6 +80,14 @@ public class WebSocketResponse {
* entries (nameLen + name + seqTxn).
*/
public static final byte STATUS_DURABLE_ACK = 0x02;
+ /**
+ * Per-table local-durability acknowledgment. Emitted when the
+ * connection's granted durable-ack tier set includes {@code local}, as
+ * the tables' sequencer records are fdatasync'd on the server -- the
+ * acked transactions survive power loss, though not the loss of the
+ * server's disk. Same payload layout as {@link #STATUS_DURABLE_ACK}.
+ */
+ public static final byte STATUS_LOCAL_DURABLE_ACK = 0x0E;
public static final byte STATUS_INTERNAL_ERROR = 0x06;
/**
* Node cannot serve writes (read-only replica / demoting primary). Reserved:
@@ -127,6 +135,19 @@ public static WebSocketResponse durableAck(String tableName, long seqTxn) {
return response;
}
+ /**
+ * Creates a local-durability ACK response with a single table entry.
+ */
+ @TestOnly
+ public static WebSocketResponse localDurableAck(String tableName, long seqTxn) {
+ WebSocketResponse response = new WebSocketResponse();
+ response.status = STATUS_LOCAL_DURABLE_ACK;
+ response.sequence = -1;
+ response.tableNames.add(tableName);
+ response.tableSeqTxns.add(seqTxn);
+ return response;
+ }
+
/**
* Creates an error response.
*/
@@ -160,7 +181,7 @@ public static boolean isStructurallyValid(long ptr, int length) {
return validateTableEntries(ptr + 9, length - 9);
}
- if (status == STATUS_DURABLE_ACK) {
+ if (isDurableAckStatus(status)) {
if (length < MIN_DURABLE_ACK_SIZE) {
return false;
}
@@ -218,6 +239,8 @@ public String getStatusName() {
return "OK";
case STATUS_DURABLE_ACK:
return "DURABLE_ACK";
+ case STATUS_LOCAL_DURABLE_ACK:
+ return "LOCAL_DURABLE_ACK";
case STATUS_PARSE_ERROR:
return "PARSE_ERROR";
case STATUS_SCHEMA_MISMATCH:
@@ -256,6 +279,20 @@ public boolean isDurableAck() {
return status == STATUS_DURABLE_ACK;
}
+ /**
+ * Returns true when this is a per-table local-durability ACK
+ * (STATUS_LOCAL_DURABLE_ACK).
+ */
+ public boolean isLocalDurableAck() {
+ return status == STATUS_LOCAL_DURABLE_ACK;
+ }
+
+ // Both durable-ack statuses share the sequence-less payload layout:
+ // status + tableCount + per-table entries.
+ private static boolean isDurableAckStatus(byte status) {
+ return status == STATUS_DURABLE_ACK || status == STATUS_LOCAL_DURABLE_ACK;
+ }
+
/**
* Returns true if this is a success response (STATUS_OK).
*/
@@ -290,7 +327,7 @@ public boolean readFrom(long ptr, int length) {
return readTableEntries(ptr + 9, length - 9);
}
- if (status == STATUS_DURABLE_ACK) {
+ if (isDurableAckStatus(status)) {
if (length < MIN_DURABLE_ACK_SIZE) {
return false;
}
@@ -333,7 +370,7 @@ public int serializedSize() {
if (status == STATUS_OK) {
return MIN_OK_RESPONSE_SIZE + tableEntriesSize();
}
- if (status == STATUS_DURABLE_ACK) {
+ if (isDurableAckStatus(status)) {
return MIN_DURABLE_ACK_SIZE + tableEntriesSize();
}
return MIN_ERROR_RESPONSE_SIZE + getErrorMessageUtf8Length();
@@ -343,8 +380,8 @@ public int serializedSize() {
public String toString() {
if (isSuccess()) {
return "WebSocketResponse{status=OK, seq=" + sequence + ", tables=" + tableNames.size() + "}";
- } else if (isDurableAck()) {
- return "WebSocketResponse{status=DURABLE_ACK, tables=" + tableNames.size() + "}";
+ } else if (isDurableAck() || isLocalDurableAck()) {
+ return "WebSocketResponse{status=" + getStatusName() + ", tables=" + tableNames.size() + "}";
} else {
return "WebSocketResponse{status=" + getStatusName() + ", seq=" + sequence +
", error=" + errorMessage + "}";
@@ -368,7 +405,7 @@ public int writeTo(long ptr) {
Unsafe.getUnsafe().putLong(ptr + offset, sequence);
offset += 8;
offset += writeTableEntries(ptr + offset);
- } else if (status == STATUS_DURABLE_ACK) {
+ } else if (isDurableAckStatus(status)) {
offset += writeTableEntries(ptr + offset);
} else {
Unsafe.getUnsafe().putLong(ptr + offset, sequence);
@@ -482,7 +519,7 @@ private int writeTableEntries(long ptr) {
}
private int getErrorMessageUtf8Length() {
- if (status == STATUS_OK || status == STATUS_DURABLE_ACK || errorMessage == null || errorMessage.isEmpty()) {
+ if (status == STATUS_OK || isDurableAckStatus(status) || errorMessage == null || errorMessage.isEmpty()) {
errorMessageUtf8Length = 0;
return 0;
}
diff --git a/core/src/main/java/io/questdb/client/cutlass/qwp/client/sf/cursor/BackgroundDrainer.java b/core/src/main/java/io/questdb/client/cutlass/qwp/client/sf/cursor/BackgroundDrainer.java
index facd872bb..57bc08f3a 100644
--- a/core/src/main/java/io/questdb/client/cutlass/qwp/client/sf/cursor/BackgroundDrainer.java
+++ b/core/src/main/java/io/questdb/client/cutlass/qwp/client/sf/cursor/BackgroundDrainer.java
@@ -28,6 +28,7 @@
import io.questdb.client.SenderErrorHandler;
import io.questdb.client.cutlass.http.client.WebSocketClient;
import io.questdb.client.cutlass.http.client.WebSocketUpgradeException;
+import io.questdb.client.cutlass.qwp.client.DurableAckTiers;
import io.questdb.client.cutlass.qwp.client.QwpAuthFailedException;
import io.questdb.client.cutlass.qwp.client.QwpCredentialUnavailableException;
import io.questdb.client.cutlass.qwp.client.QwpDurableAckMismatchException;
@@ -174,7 +175,7 @@ public final class BackgroundDrainer implements Runnable {
private final long reconnectInitialBackoffMillis;
private final long reconnectMaxBackoffMillis;
private final long reconnectMaxDurationMillis;
- private final boolean requestDurableAck;
+ private final int durableAckTiers;
private final long segmentSizeBytes;
private final long sfMaxTotalBytes;
private final String slotPath;
@@ -267,12 +268,12 @@ public BackgroundDrainer(
long reconnectMaxDurationMillis,
long reconnectInitialBackoffMillis,
long reconnectMaxBackoffMillis,
- boolean requestDurableAck,
+ int durableAckTiers,
long durableAckKeepaliveIntervalMillis
) {
this(slotPath, segmentSizeBytes, sfMaxTotalBytes, clientFactory,
reconnectMaxDurationMillis, reconnectInitialBackoffMillis,
- reconnectMaxBackoffMillis, requestDurableAck,
+ reconnectMaxBackoffMillis, durableAckTiers,
durableAckKeepaliveIntervalMillis,
CursorWebSocketSendLoop.DEFAULT_MAX_HEAD_FRAME_REJECTIONS,
CursorWebSocketSendLoop.DEFAULT_POISON_MIN_ESCALATION_WINDOW_MILLIS,
@@ -293,7 +294,7 @@ public BackgroundDrainer(
long reconnectMaxDurationMillis,
long reconnectInitialBackoffMillis,
long reconnectMaxBackoffMillis,
- boolean requestDurableAck,
+ int durableAckTiers,
long durableAckKeepaliveIntervalMillis,
int maxHeadFrameRejections,
long poisonMinEscalationWindowMillis,
@@ -301,7 +302,7 @@ public BackgroundDrainer(
) {
this(slotPath, segmentSizeBytes, sfMaxTotalBytes, 0L, clientFactory,
reconnectMaxDurationMillis, reconnectInitialBackoffMillis,
- reconnectMaxBackoffMillis, requestDurableAck,
+ reconnectMaxBackoffMillis, durableAckTiers,
durableAckKeepaliveIntervalMillis, maxHeadFrameRejections,
poisonMinEscalationWindowMillis,
catchUpCapGapMinEscalationWindowMillis);
@@ -320,7 +321,7 @@ public BackgroundDrainer(
long reconnectMaxDurationMillis,
long reconnectInitialBackoffMillis,
long reconnectMaxBackoffMillis,
- boolean requestDurableAck,
+ int durableAckTiers,
long durableAckKeepaliveIntervalMillis,
int maxHeadFrameRejections,
long poisonMinEscalationWindowMillis,
@@ -334,7 +335,7 @@ public BackgroundDrainer(
this.reconnectMaxDurationMillis = reconnectMaxDurationMillis;
this.reconnectInitialBackoffMillis = reconnectInitialBackoffMillis;
this.reconnectMaxBackoffMillis = reconnectMaxBackoffMillis;
- this.requestDurableAck = requestDurableAck;
+ this.durableAckTiers = durableAckTiers;
this.durableAckKeepaliveIntervalMillis = durableAckKeepaliveIntervalMillis;
this.maxHeadFrameRejections = maxHeadFrameRejections;
this.poisonMinEscalationWindowMillis = poisonMinEscalationWindowMillis;
@@ -350,7 +351,7 @@ public BackgroundDrainer(
*/
@TestOnly
public BackgroundDrainer() {
- this(null, 0L, 0L, null, 0L, 0L, 0L, false, 0L,
+ this(null, 0L, 0L, null, 0L, 0L, 0L, DurableAckTiers.NONE, 0L,
CursorWebSocketSendLoop.DEFAULT_MAX_HEAD_FRAME_REJECTIONS, 0L, 0L);
}
@@ -1135,7 +1136,7 @@ public void run() {
clientFactory,
reconnectInitialBackoffMillis,
reconnectMaxBackoffMillis,
- requestDurableAck,
+ durableAckTiers,
durableAckKeepaliveIntervalMillis,
maxHeadFrameRejections,
poisonMinEscalationWindowMillis,
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 6643bf036..e01cbcfff 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
@@ -34,6 +34,7 @@
import io.questdb.client.cutlass.qwp.client.NativeBufferWriter;
import io.questdb.client.cutlass.qwp.client.QwpAuthFailedException;
import io.questdb.client.cutlass.qwp.client.QwpCredentialUnavailableException;
+import io.questdb.client.cutlass.qwp.client.DurableAckTiers;
import io.questdb.client.cutlass.qwp.client.QwpDurableAckMismatchException;
import io.questdb.client.cutlass.qwp.client.QwpIngressRoleRejectedException;
import io.questdb.client.cutlass.qwp.client.QwpRoleMismatchException;
@@ -256,6 +257,11 @@ public final class CursorWebSocketSendLoop implements QuietCloseable {
// (default), the loop trims on OK as it always has and ignores any
// STATUS_DURABLE_ACK frames that might still arrive (logs a warning).
private final boolean durableAckMode;
+ // True when the local tier is requested without the replicated one: the
+ // trim trigger is then STATUS_LOCAL_DURABLE_ACK. Any request including
+ // the replicated tier trims on STATUS_DURABLE_ACK (strongest requested
+ // wins) and treats local acks as progress signals only.
+ private final boolean isLocalAckTrimming;
// Per-table cumulative durable-upload watermarks, populated only when
// durableAckMode is true. Updated from STATUS_DURABLE_ACK frame entries
// (each entry is monotonically non-decreasing per spec). Reset on every
@@ -263,6 +269,11 @@ public final class CursorWebSocketSendLoop implements QuietCloseable {
// by the server -- holding stale watermarks across the wire boundary
// would falsely advance trim before re-confirmation.
private final CharSequenceLongHashMap durableTableWatermarks = new CharSequenceLongHashMap();
+ // Per-table local-fsync watermarks from STATUS_LOCAL_DURABLE_ACK frames
+ // when BOTH tiers are requested. Progress observability only -- the trim
+ // path never reads it (the replicated ack drives the trim). In local-only
+ // mode local acks feed durableTableWatermarks directly instead.
+ private final CharSequenceLongHashMap localDurableTableWatermarks = new CharSequenceLongHashMap();
// Pre-converted to nanos. Consulted only by the orphan terminal policy. Zero disables
// the dwell entirely (count-only escalation at MAX_CATCHUP_CAP_GAP_ATTEMPTS); the
// user-facing 5-minute default is applied at the config layer.
@@ -299,6 +310,7 @@ public final class CursorWebSocketSendLoop implements QuietCloseable {
// Counters for observability of the durable-ack path. Both are zero
// when durableAckMode is false.
private final AtomicLong totalDurableAcks = new AtomicLong();
+ private final AtomicLong totalLocalDurableAcks = new AtomicLong();
private final AtomicLong totalDurableTrimAdvances = new AtomicLong();
// Cumulative count of frames the loop has re-sent during post-reconnect
// catch-up windows. Bumped once per frame on every iteration that
@@ -603,13 +615,14 @@ public CursorWebSocketSendLoop(WebSocketClient client, CursorSendEngine engine,
long reconnectMaxBackoffMillis) {
this(client, engine, fsnAtZero, parkNanos, reconnectFactory,
reconnectInitialBackoffMillis,
- reconnectMaxBackoffMillis, false);
+ reconnectMaxBackoffMillis, DurableAckTiers.NONE);
}
/**
* Same as the seven-arg constructor but with explicit control over
- * durable-ack-driven trim. {@code durableAckMode = true} switches the loop
- * to trim only on {@link WebSocketResponse#STATUS_DURABLE_ACK} frames; OK
+ * durable-ack-driven trim. A non-empty {@code durableAckTiers} set
+ * ({@link DurableAckTiers}) switches the loop to trim only on the
+ * strongest requested tier's ack frames; OK
* frames are queued until their per-table seqTxns are covered by a durable
* watermark. The default (false) preserves the historical OK-driven trim
* and ignores any durable-ack frames that arrive (logging a warning, since
@@ -620,10 +633,10 @@ public CursorWebSocketSendLoop(WebSocketClient client, CursorSendEngine engine,
ReconnectFactory reconnectFactory,
long reconnectInitialBackoffMillis,
long reconnectMaxBackoffMillis,
- boolean durableAckMode) {
+ int durableAckTiers) {
this(client, engine, fsnAtZero, parkNanos, reconnectFactory,
reconnectInitialBackoffMillis,
- reconnectMaxBackoffMillis, durableAckMode,
+ reconnectMaxBackoffMillis, durableAckTiers,
DEFAULT_DURABLE_ACK_KEEPALIVE_INTERVAL_MILLIS);
}
@@ -639,11 +652,11 @@ public CursorWebSocketSendLoop(WebSocketClient client, CursorSendEngine engine,
ReconnectFactory reconnectFactory,
long reconnectInitialBackoffMillis,
long reconnectMaxBackoffMillis,
- boolean durableAckMode,
+ int durableAckTiers,
long durableAckKeepaliveIntervalMillis) {
this(client, engine, fsnAtZero, parkNanos, reconnectFactory,
reconnectInitialBackoffMillis,
- reconnectMaxBackoffMillis, durableAckMode,
+ reconnectMaxBackoffMillis, durableAckTiers,
durableAckKeepaliveIntervalMillis, DEFAULT_MAX_HEAD_FRAME_REJECTIONS);
}
@@ -661,12 +674,12 @@ public CursorWebSocketSendLoop(WebSocketClient client, CursorSendEngine engine,
ReconnectFactory reconnectFactory,
long reconnectInitialBackoffMillis,
long reconnectMaxBackoffMillis,
- boolean durableAckMode,
+ int durableAckTiers,
long durableAckKeepaliveIntervalMillis,
int maxHeadFrameRejections) {
this(client, engine, fsnAtZero, parkNanos, reconnectFactory,
reconnectInitialBackoffMillis,
- reconnectMaxBackoffMillis, durableAckMode,
+ reconnectMaxBackoffMillis, durableAckTiers,
durableAckKeepaliveIntervalMillis, maxHeadFrameRejections, 0L);
}
@@ -681,13 +694,13 @@ public CursorWebSocketSendLoop(WebSocketClient client, CursorSendEngine engine,
ReconnectFactory reconnectFactory,
long reconnectInitialBackoffMillis,
long reconnectMaxBackoffMillis,
- boolean durableAckMode,
+ int durableAckTiers,
long durableAckKeepaliveIntervalMillis,
int maxHeadFrameRejections,
long poisonMinEscalationWindowMillis) {
this(client, engine, fsnAtZero, parkNanos, reconnectFactory,
reconnectInitialBackoffMillis,
- reconnectMaxBackoffMillis, durableAckMode,
+ reconnectMaxBackoffMillis, durableAckTiers,
durableAckKeepaliveIntervalMillis, maxHeadFrameRejections,
poisonMinEscalationWindowMillis, 0L);
}
@@ -714,14 +727,14 @@ public CursorWebSocketSendLoop(WebSocketClient client, CursorSendEngine engine,
ReconnectFactory reconnectFactory,
long reconnectInitialBackoffMillis,
long reconnectMaxBackoffMillis,
- boolean durableAckMode,
+ int durableAckTiers,
long durableAckKeepaliveIntervalMillis,
int maxHeadFrameRejections,
long poisonMinEscalationWindowMillis,
long catchUpCapGapMinEscalationWindowMillis) {
this(client, engine, fsnAtZero, parkNanos, reconnectFactory,
reconnectInitialBackoffMillis,
- reconnectMaxBackoffMillis, durableAckMode,
+ reconnectMaxBackoffMillis, durableAckTiers,
durableAckKeepaliveIntervalMillis, maxHeadFrameRejections,
poisonMinEscalationWindowMillis, catchUpCapGapMinEscalationWindowMillis,
CatchUpCapGapPolicy.RETRY_FOREVER);
@@ -737,7 +750,7 @@ private CursorWebSocketSendLoop(WebSocketClient client, CursorSendEngine engine,
ReconnectFactory reconnectFactory,
long reconnectInitialBackoffMillis,
long reconnectMaxBackoffMillis,
- boolean durableAckMode,
+ int durableAckTiers,
long durableAckKeepaliveIntervalMillis,
int maxHeadFrameRejections,
long poisonMinEscalationWindowMillis,
@@ -898,7 +911,8 @@ private CursorWebSocketSendLoop(WebSocketClient client, CursorSendEngine engine,
this.reconnectFactory = reconnectFactory;
this.reconnectInitialBackoffMillis = reconnectInitialBackoffMillis;
this.reconnectMaxBackoffMillis = reconnectMaxBackoffMillis;
- this.durableAckMode = durableAckMode;
+ this.durableAckMode = durableAckTiers != DurableAckTiers.NONE;
+ this.isLocalAckTrimming = DurableAckTiers.isTrimOnLocalAck(durableAckTiers);
// Saturate, never multiply raw -- the same hazard the cap-gap dwell above
// guards. A raw multiply wraps a large millisecond value NEGATIVE, and both of
// these read as "elapsed >= window", so a negative makes the gate trivially
@@ -941,7 +955,7 @@ public CursorWebSocketSendLoop(WebSocketClient client, CursorSendEngine engine,
ReconnectFactory reconnectFactory,
long reconnectInitialBackoffMillis,
long reconnectMaxBackoffMillis,
- boolean durableAckMode,
+ int durableAckTiers,
long durableAckKeepaliveIntervalMillis,
int maxHeadFrameRejections,
long poisonMinEscalationWindowMillis,
@@ -949,7 +963,7 @@ public CursorWebSocketSendLoop(WebSocketClient client, CursorSendEngine engine,
ReconnectPolicy reconnectPolicy) {
this(client, engine, fsnAtZero, parkNanos, reconnectFactory,
reconnectInitialBackoffMillis,
- reconnectMaxBackoffMillis, durableAckMode,
+ reconnectMaxBackoffMillis, durableAckTiers,
durableAckKeepaliveIntervalMillis, maxHeadFrameRejections,
poisonMinEscalationWindowMillis, catchUpCapGapMinEscalationWindowMillis,
catchUpPolicyFor(reconnectPolicy));
@@ -1477,6 +1491,16 @@ public long getTotalDurableAcks() {
return totalDurableAcks.get();
}
+ /**
+ * Total {@code STATUS_LOCAL_DURABLE_ACK} frames received since the loop
+ * started. Always 0 unless the local tier was requested. In local-only
+ * mode these frames drive the trim; with both tiers requested they are
+ * progress signals and the count grows independently of trims.
+ */
+ public long getTotalLocalDurableAcks() {
+ return totalLocalDurableAcks.get();
+ }
+
/**
* Total times a durable-ack frame caused {@link CursorSendEngine#acknowledge}
* to advance. Always 0 when {@code durableAckMode} is false. A non-zero
@@ -1680,6 +1704,23 @@ private void applyDurableAck() {
drainPendingDurable();
}
+ /**
+ * Records per-table local-fsync watermarks from a
+ * STATUS_LOCAL_DURABLE_ACK frame when both tiers are requested. Progress
+ * observability only: the trim path is driven by the replicated ack.
+ */
+ private void applyLocalDurableAckProgress() {
+ int n = response.getTableEntryCount();
+ for (int i = 0; i < n; i++) {
+ String name = response.getTableName(i);
+ long seqTxn = response.getTableSeqTxn(i);
+ long current = localDurableTableWatermarks.get(name);
+ if (seqTxn > current) {
+ localDurableTableWatermarks.put(name, seqTxn);
+ }
+ }
+ }
+
/**
* Drives the very first connect attempt on the I/O thread, used in the
* async-initial-connect mode (constructed with {@code client == null}).
@@ -1704,6 +1745,7 @@ private void clearDurableAckTracking() {
releasePendingEntry(pendingDurable.pollFirst());
}
durableTableWatermarks.clear();
+ localDurableTableWatermarks.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.
@@ -3932,6 +3974,24 @@ public void onBinaryMessage(long payloadPtr, int payloadLen) {
applyDurableAck();
return;
}
+ if (response.isLocalDurableAck()) {
+ if (!durableAckMode) {
+ LOG.warn("received STATUS_LOCAL_DURABLE_ACK frame without opt-in -- ignoring");
+ return;
+ }
+ totalLocalDurableAcks.incrementAndGet();
+ if (isLocalAckTrimming) {
+ // Local is the strongest requested tier, so its ack is
+ // this connection's trim trigger -- same watermark and
+ // drain path the replicated ack drives otherwise.
+ applyDurableAck();
+ } else {
+ // Both tiers requested: the replicated ack trims; the
+ // local ack is an early progress signal only.
+ applyLocalDurableAckProgress();
+ }
+ return;
+ }
// Application-layer rejection by the server. Classify by status
// byte → SenderError.Category, resolve policy (default mapping
// for now; user-override resolution lands in a later commit),
diff --git a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/DeltaDictRecoveryTest.java b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/DeltaDictRecoveryTest.java
index 66ba07497..fbc833d86 100644
--- a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/DeltaDictRecoveryTest.java
+++ b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/DeltaDictRecoveryTest.java
@@ -24,6 +24,7 @@
package io.questdb.client.test.cutlass.qwp.client;
+import io.questdb.client.cutlass.qwp.client.DurableAckTiers;
import io.questdb.client.Sender;
import io.questdb.client.SenderError;
import io.questdb.client.cutlass.line.LineSenderException;
@@ -345,7 +346,7 @@ public void testQueuedOrphanCannotAdoptSlotWhileQuarantineRecreatesItsName() thr
BackgroundDrainer queuedDrainer = new BackgroundDrainer(
staleSnapshotPath, 256, 8192, () -> null,
- 1000, 1, 10, true, 0);
+ 1000, 1, 10, DurableAckTiers.REPLICATED, 0);
Thread drainerThread = new Thread(queuedDrainer, "qwp-queued-orphan");
drainerThread.start();
drainerThread.join(5_000);
@@ -982,7 +983,7 @@ public void testPersistFailureSurfacesAsLineSenderException() throws Exception {
slot, 4L * 1024 * 1024, CursorSendEngine.DEFAULT_APPEND_DEADLINE_NANOS,
CursorSendEngine.DEFAULT_APPEND_DEADLINE_NANOS, ff);
Sender sender = QwpWebSocketSender.connect(
- "localhost", port, null, 0, 0, 0L, null, false, engine);
+ "localhost", port, null, 0, 0, 0L, null, DurableAckTiers.NONE, engine);
try {
ff.armed = true; // the next dictionary append cannot grow its window
sender.table("m").symbol("s", "boom").longColumn("v", 1L).atNow();
@@ -1552,7 +1553,7 @@ public void testFullDictFramesRecoverBesideASurvivingPopulatedDictionary() throw
slot.toString(), 4L * 1024 * 1024, CursorSendEngine.DEFAULT_APPEND_DEADLINE_NANOS,
CursorSendEngine.DEFAULT_APPEND_DEADLINE_NANOS, phase1DictFf);
try (Sender s1 = QwpWebSocketSender.connect(
- "localhost", port, null, 0, 0, 0L, null, false, phase1Engine, 0L)) {
+ "localhost", port, null, 0, 0, 0L, null, DurableAckTiers.NONE, phase1Engine, 0L)) {
for (int i = 0; i < DISTINCT_SYMBOLS; i++) {
s1.table("m").symbol("s", "sym-" + i).longColumn("v", i).atNow();
s1.flush();
@@ -1639,7 +1640,7 @@ public void testDiscardedSurvivingDictionaryIsUnlinked() throws Exception {
slot.toString(), 4L * 1024 * 1024, CursorSendEngine.DEFAULT_APPEND_DEADLINE_NANOS,
CursorSendEngine.DEFAULT_APPEND_DEADLINE_NANOS, new UnopenableDictFacade());
try (Sender s1 = QwpWebSocketSender.connect(
- "localhost", port, null, 0, 0, 0L, null, false, phase1Engine, 0L)) {
+ "localhost", port, null, 0, 0, 0L, null, DurableAckTiers.NONE, phase1Engine, 0L)) {
for (int i = 0; i < DISTINCT_SYMBOLS; i++) {
s1.table("m").symbol("s", "sym-" + i).longColumn("v", i).atNow();
s1.flush();
@@ -1720,7 +1721,7 @@ public void testFullDictFramesRecoverInFullDictModeInsteadOfBricking() throws Ex
slot.toString(), 4L * 1024 * 1024, CursorSendEngine.DEFAULT_APPEND_DEADLINE_NANOS,
CursorSendEngine.DEFAULT_APPEND_DEADLINE_NANOS, phase1DictFf);
try (Sender s1 = QwpWebSocketSender.connect(
- "localhost", port, null, 0, 0, 0L, null, false, phase1Engine, 0L)) {
+ "localhost", port, null, 0, 0, 0L, null, DurableAckTiers.NONE, phase1Engine, 0L)) {
for (int i = 0; i < DISTINCT_SYMBOLS; i++) {
s1.table("m").symbol("s", "sym-" + i).longColumn("v", i).atNow();
s1.flush();
@@ -2119,7 +2120,7 @@ public void testPersistFailureDegradesToFullDictInsteadOfKillingFlushForever() t
slot, 4L * 1024 * 1024, CursorSendEngine.DEFAULT_APPEND_DEADLINE_NANOS,
CursorSendEngine.DEFAULT_APPEND_DEADLINE_NANOS, ff);
Sender sender = QwpWebSocketSender.connect(
- "localhost", port, null, 0, 0, 0L, null, false, engine);
+ "localhost", port, null, 0, 0, 0L, null, DurableAckTiers.NONE, engine);
try {
// Armed from the start, so the very first ensureAppendMap is refused --
// a later append would sit inside the window already mapped and never
@@ -2354,7 +2355,7 @@ public void testTransientDictFaultRecoversInFullDictModeWhenEveryFrameIsSelfSuff
slot.toString(), 4L * 1024 * 1024, CursorSendEngine.DEFAULT_APPEND_DEADLINE_NANOS,
CursorSendEngine.DEFAULT_APPEND_DEADLINE_NANOS, new UnopenableDictFacade());
try (Sender s1 = QwpWebSocketSender.connect(
- "localhost", port, null, 0, 0, 0L, null, false, phase1Engine, 0L)) {
+ "localhost", port, null, 0, 0, 0L, null, DurableAckTiers.NONE, phase1Engine, 0L)) {
for (int i = 0; i < DISTINCT_SYMBOLS; i++) {
s1.table("m").symbol("s", "sym-" + i).longColumn("v", i).atNow();
s1.flush();
diff --git a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/MmapFaultDegradesTest.java b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/MmapFaultDegradesTest.java
index 360cbf5e9..6e0bf773f 100644
--- a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/MmapFaultDegradesTest.java
+++ b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/MmapFaultDegradesTest.java
@@ -24,6 +24,7 @@
package io.questdb.client.test.cutlass.qwp.client;
+import io.questdb.client.cutlass.qwp.client.DurableAckTiers;
import io.questdb.client.Sender;
import io.questdb.client.cutlass.line.LineSenderException;
import io.questdb.client.cutlass.qwp.client.QwpWebSocketSender;
@@ -101,7 +102,7 @@ public void testMmapAccessFaultDegradesPersistInsteadOfPropagating() throws Exce
slot, 4L * 1024 * 1024, 64L * 1024 * 1024,
CursorSendEngine.DEFAULT_APPEND_DEADLINE_NANOS, ff);
QwpWebSocketSender sender = QwpWebSocketSender.connect(
- "localhost", port, null, 0, 0, 0L, null, false, engine);
+ "localhost", port, null, 0, 0, 0L, null, DurableAckTiers.NONE, engine);
try {
ff.armed = true; // the next dictionary mmap growth raises the fault
sender.table("m").symbol("s", "boom").longColumn("v", 1L).atNow();
@@ -190,7 +191,7 @@ public void testMmapAccessFaultDuringDictHealDegradesInsteadOfEscapingBuild() th
// now so ONLY the heal's later MAP_RW growth of that same fd raises the fault.
ff.armed = true;
QwpWebSocketSender resumed = QwpWebSocketSender.connect(
- "localhost", port, null, 0, 0, 0L, null, false, engine);
+ "localhost", port, null, 0, 0, 0L, null, DurableAckTiers.NONE, engine);
try {
Assert.assertFalse("a recognised mmap access fault during the dictionary heal "
+ "must degrade the sender to self-sufficient frames, not escape build()",
diff --git a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/QwpWebSocketSenderJvmErrorCleanupTest.java b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/QwpWebSocketSenderJvmErrorCleanupTest.java
index d0eb80db0..0eee60638 100644
--- a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/QwpWebSocketSenderJvmErrorCleanupTest.java
+++ b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/QwpWebSocketSenderJvmErrorCleanupTest.java
@@ -24,6 +24,7 @@
package io.questdb.client.test.cutlass.qwp.client;
+import io.questdb.client.cutlass.qwp.client.DurableAckTiers;
import io.questdb.client.cutlass.http.client.WebSocketClient;
import io.questdb.client.cutlass.line.LineSenderException;
import io.questdb.client.cutlass.qwp.client.QwpHostHealthTracker;
@@ -364,7 +365,7 @@ public void setQwpMaxVersion(int maxVersion) {
}
@Override
- public void setQwpRequestDurableAck(boolean enabled) {
+ public void setQwpDurableAckTiers(int tiers) {
}
@Override
diff --git a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/QwpWebSocketSenderTest.java b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/QwpWebSocketSenderTest.java
index 3911f5c05..d90fd478a 100644
--- a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/QwpWebSocketSenderTest.java
+++ b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/QwpWebSocketSenderTest.java
@@ -24,6 +24,7 @@
package io.questdb.client.test.cutlass.qwp.client;
+import io.questdb.client.cutlass.qwp.client.DurableAckTiers;
import io.questdb.client.cutlass.line.LineSenderException;
import io.questdb.client.cutlass.line.array.DoubleArray;
import io.questdb.client.cutlass.line.array.LongArray;
@@ -349,7 +350,7 @@ public void testFlushAppendFailureDoesNotLeaveMicrobatchBufferInUse() throws Exc
CursorSendEngine engine = new CursorSendEngine(null, minSegmentBytes, minSegmentBytes, 1L);
try (QwpWebSocketSender sender = QwpWebSocketSender.connect(
"localhost", port, null, Integer.MAX_VALUE, 0, 0L, null,
- false, engine, 0L)) {
+ DurableAckTiers.NONE, engine, 0L)) {
sender.table("t").longColumn("v", 1L).atNow();
try {
diff --git a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/SelfSufficientFramesTest.java b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/SelfSufficientFramesTest.java
index e305210b1..1d1aaebfa 100644
--- a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/SelfSufficientFramesTest.java
+++ b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/SelfSufficientFramesTest.java
@@ -24,6 +24,7 @@
package io.questdb.client.test.cutlass.qwp.client;
+import io.questdb.client.cutlass.qwp.client.DurableAckTiers;
import io.questdb.client.Sender;
import io.questdb.client.cutlass.line.LineSenderException;
import io.questdb.client.cutlass.qwp.client.QwpWebSocketSender;
@@ -159,7 +160,7 @@ public void testDiskModeFallsBackToFullDictWhenPersistedDictUnopenable() throws
slot, 4L * 1024 * 1024, CursorSendEngine.DEFAULT_APPEND_DEADLINE_NANOS,
CursorSendEngine.DEFAULT_APPEND_DEADLINE_NANOS, dictFf);
try (Sender sender = QwpWebSocketSender.connect(
- "localhost", port, null, 0, 0, 0L, null, false, engine)) {
+ "localhost", port, null, 0, 0, 0L, null, DurableAckTiers.NONE, engine)) {
sender.table("foo").symbol("s", "alpha").longColumn("v", 1L).atNow();
sender.flush();
waitFor(() -> handler.batches.size() >= 1, 5_000);
@@ -751,7 +752,7 @@ public void testDictionaryLargerThanTheCapShipsAsChunkedDictionaryFrames() throw
// 40 x ~60-byte entries is ~2.4 KB of dictionary against a 512-byte cap,
// so no single frame can carry it and at least five chunks are required.
try (Sender sender = QwpWebSocketSender.connect(
- "localhost", port, null, 1_000_000, 0, 0L, null, false, engine)) {
+ "localhost", port, null, 1_000_000, 0, 0L, null, DurableAckTiers.NONE, engine)) {
String pad = TestUtils.repeat("x", 55);
for (int i = 0; i < symbols; i++) {
String sym = String.format("%04d", i) + pad;
@@ -826,7 +827,7 @@ public void testFullDictNearCapFallsBackToChunkedDictionary() throws Exception {
CursorSendEngine.DEFAULT_APPEND_DEADLINE_NANOS, dictFf);
List