* This setting is only supported for WebSocket transport. * @@ -2703,10 +2706,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;
}
@@ -3695,7 +3736,7 @@ private QwpWebSocketSender buildWebSocket(HttpTokenProvider tokenProvider) {
actualAutoFlushBytes,
actualAutoFlushIntervalNanos,
wsAuthHeader,
- requestDurableAck,
+ durableAckTiers,
cursorEngine,
actualCloseFlushTimeoutMillis,
actualReconnectMaxDurationMillis,
@@ -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");
@@ -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) {
@@ -4772,7 +4809,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 "true"
+ // request token and expect the "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 {@code "enabled"} token; explicit tier 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} maps to the replicated tier with
+ * {@link #LEGACY_TRUE} set; {@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 literal {@code "true"},
+ * the only request value servers without tier support recognize.
+ */
+ 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/QwpDurableAckMismatchException.java b/core/src/main/java/io/questdb/client/cutlass/qwp/client/QwpDurableAckMismatchException.java
index b20ae250..e8a58e5d 100644
--- a/core/src/main/java/io/questdb/client/cutlass/qwp/client/QwpDurableAckMismatchException.java
+++ b/core/src/main/java/io/questdb/client/cutlass/qwp/client/QwpDurableAckMismatchException.java
@@ -25,27 +25,49 @@
package io.questdb.client.cutlass.qwp.client;
import io.questdb.client.cutlass.http.client.HttpClientException;
+import org.jetbrains.annotations.Nullable;
/**
- * The client opted in via {@code request_durable_ack=on} but the server cannot
- * honour it. Terminal: retrying the same configured endpoints will not turn a
- * non-primary node into a durable-ack-capable primary, so the connect path
- * fails fast instead of burning the full reconnect budget.
+ * The client requested durable-ack tiers via {@code request_durable_ack}
+ * ({@code local}, {@code replicated}, {@code local,replicated}, or the legacy
+ * {@code on}) but the server did not grant that exact set. Terminal: retrying
+ * the same configured endpoints will not turn a server that cannot grant the
+ * set into one that can, so the connect path fails fast instead of burning the
+ * full reconnect budget.
*
- * Two flavours collapse into this exception:
- *
- * STATUS_DURABLE_ACK response format:
+ * STATUS_DURABLE_ACK / STATUS_LOCAL_DURABLE_ACK response format:
*
+ * The {@link BackgroundDrainer} hands its configured tier set to every {@link CursorWebSocketSendLoop} it
+ * builds. With LOCAL alone, {@code STATUS_LOCAL_DURABLE_ACK} is the trim trigger: STATUS_OK must not trim
+ * the slot, and each local ack must advance the drainer's watermark until the slot is fully drained. A
+ * drainer that passed a different tier set would either trim on STATUS_OK (no tier) or treat local acks
+ * as progress only and never finish (replicated tier).
+ *
+ * Wire realism: a real {@link TestWebSocketServer} grants the requested {@code local} tier on upgrade,
+ * OKs every frame at once and sends the local acks only when the test releases them.
+ */
+public class BackgroundDrainerLocalDurableAckTest {
+
+ private static final long FAST_BACKOFF_MAX_MILLIS = 4L;
+ private static final long FAST_BACKOFF_MILLIS = 1L;
+ private static final long RECONNECT_MAX_DURATION_MILLIS = 60_000L;
+ private static final int SEEDED_FRAMES = 5;
+ private static final long SEGMENT_SIZE_BYTES = 16384L;
+ private static final long SF_MAX_TOTAL_BYTES = 1L << 20;
+ private static final String TABLE = "trades";
+
+ private String slotPath;
+
+ @Before
+ public void setUp() {
+ slotPath = Paths.get(System.getProperty("java.io.tmpdir"),
+ "qdb-drainer-local-ack-" + System.nanoTime()).toString();
+ assertEquals("mkdir slot dir", 0, Files.mkdir(slotPath, Files.DIR_MODE_DEFAULT));
+ }
+
+ @After
+ public void tearDown() {
+ rmDirRec(slotPath);
+ }
+
+ @Test
+ public void testOrphanSlotWithLocalTierTrimsOnLocalDurableAcks() throws Exception {
+ TestUtils.assertMemoryLeak(() -> {
+ seedSlot();
+ LocalAckHoldingHandler handler = new LocalAckHoldingHandler();
+ try (TestWebSocketServer server = new TestWebSocketServer(handler, true)) {
+ server.start();
+ assertTrue(server.awaitStart(5, TimeUnit.SECONDS));
+ BackgroundDrainer drainer = new BackgroundDrainer(
+ slotPath,
+ SEGMENT_SIZE_BYTES,
+ SF_MAX_TOTAL_BYTES,
+ new LocalTierWireFactory(server.getPort()),
+ RECONNECT_MAX_DURATION_MILLIS,
+ FAST_BACKOFF_MILLIS,
+ FAST_BACKOFF_MAX_MILLIS,
+ DurableAckTiers.LOCAL,
+ /* durableAckKeepaliveIntervalMillis */ 200L);
+ Thread runner = new Thread(drainer, "test-local-ack-drainer");
+ runner.setDaemon(true);
+ runner.start();
+ try {
+ assertEquals("the server must see the local tier requested on upgrade",
+ "local", server.pollDurableAckRequest(5, TimeUnit.SECONDS));
+ assertTrue("server did not receive and OK every orphan frame",
+ handler.allFramesOked.await(5, TimeUnit.SECONDS));
+ // Give the drainer several 50ms polls to observe any OK-driven trim.
+ Thread.sleep(250);
+ assertEquals("STATUS_OK alone must not trim a LOCAL-tier orphan slot",
+ -1L, drainer.getAckedFsn());
+ assertEquals(BackgroundDrainer.DrainOutcome.PENDING, drainer.outcome());
+
+ handler.sendLocalDurableAck(1);
+ awaitAckedFsn(drainer, 1);
+ assertEquals("a local ack covering frames 0..1 must trim exactly that far",
+ 1L, drainer.getAckedFsn());
+ assertEquals(BackgroundDrainer.DrainOutcome.PENDING, drainer.outcome());
+
+ handler.sendLocalDurableAck(SEEDED_FRAMES - 1);
+ runner.join(10_000L);
+ } finally {
+ if (runner.isAlive()) {
+ drainer.requestStop();
+ runner.join(5_000L);
+ }
+ }
+
+ assertEquals("local acks must drain the orphan slot to completion",
+ BackgroundDrainer.DrainOutcome.SUCCESS, drainer.outcome());
+ assertEquals(SEEDED_FRAMES - 1L, drainer.getAckedFsn());
+ assertFalse("no .failed sentinel after a successful drain",
+ Files.exists(slotPath + "/" + OrphanScanner.FAILED_SENTINEL_NAME));
+ assertEquals("the drain must not have needed a reconnect", 1, server.handshakeCount());
+ }
+ });
+ }
+
+ private static void awaitAckedFsn(BackgroundDrainer drainer, long expected) throws InterruptedException {
+ long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(5);
+ while (drainer.getAckedFsn() < expected && System.nanoTime() < deadline) {
+ Thread.sleep(10);
+ }
+ assertTrue("STATUS_LOCAL_DURABLE_ACK did not trim the LOCAL-tier orphan slot [expected="
+ + expected + ", ackedFsn=" + drainer.getAckedFsn() + "]",
+ drainer.getAckedFsn() >= expected);
+ }
+
+ private static void rmDirRec(String dir) {
+ if (dir == null || !Files.exists(dir)) return;
+ long find = Files.findFirst(dir);
+ if (find > 0) {
+ try {
+ int rc = 1;
+ while (rc > 0) {
+ String name = Files.utf8ToString(Files.findName(find));
+ if (name != null && !".".equals(name) && !"..".equals(name)) {
+ String child = dir + "/" + name;
+ if (!Files.remove(child)) rmDirRec(child);
+ }
+ rc = Files.findNext(find);
+ }
+ } finally {
+ Files.findClose(find);
+ }
+ }
+ Files.remove(dir);
+ }
+
+ private void seedSlot() {
+ try (CursorSendEngine engine = new CursorSendEngine(slotPath, SEGMENT_SIZE_BYTES)) {
+ long buf = Unsafe.malloc(16, MemoryTag.NATIVE_DEFAULT);
+ try {
+ byte[] payload = "frame-bytes-padd".getBytes(StandardCharsets.US_ASCII);
+ for (int i = 0; i < payload.length; i++) {
+ Unsafe.getUnsafe().putByte(buf + i, payload[i]);
+ }
+ for (int i = 0; i < SEEDED_FRAMES; i++) {
+ engine.appendBlocking(buf, 16);
+ }
+ } finally {
+ Unsafe.free(buf, 16, MemoryTag.NATIVE_DEFAULT);
+ }
+ }
+ }
+
+ /**
+ * OKs every frame as it arrives (wire seq {@code n} carries table seqTxn {@code n}) and sends
+ * {@code STATUS_LOCAL_DURABLE_ACK} only when the test calls {@link #sendLocalDurableAck}.
+ */
+ private static final class LocalAckHoldingHandler implements TestWebSocketServer.WebSocketServerHandler {
+ final CountDownLatch allFramesOked = new CountDownLatch(1);
+ private TestWebSocketServer.ClientHandler client;
+ private long nextSeq;
+
+ @Override
+ public synchronized void onBinaryMessage(TestWebSocketServer.ClientHandler client, byte[] data) {
+ this.client = client;
+ long seq = nextSeq++;
+ try {
+ client.sendBinary(okFrame(seq));
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ if (nextSeq == SEEDED_FRAMES) {
+ allFramesOked.countDown();
+ }
+ }
+
+ synchronized void sendLocalDurableAck(long seqTxn) throws IOException {
+ byte[] name = TABLE.getBytes(StandardCharsets.UTF_8);
+ ByteBuffer bb = ByteBuffer.allocate(1 + 2 + 2 + name.length + 8)
+ .order(ByteOrder.LITTLE_ENDIAN);
+ bb.put(WebSocketResponse.STATUS_LOCAL_DURABLE_ACK);
+ bb.putShort((short) 1); // tableCount
+ bb.putShort((short) name.length);
+ bb.put(name);
+ bb.putLong(seqTxn);
+ client.sendBinary(bb.array());
+ }
+
+ private static byte[] okFrame(long seq) {
+ byte[] name = TABLE.getBytes(StandardCharsets.UTF_8);
+ ByteBuffer bb = ByteBuffer.allocate(1 + 8 + 2 + 2 + name.length + 8)
+ .order(ByteOrder.LITTLE_ENDIAN);
+ bb.put(WebSocketResponse.STATUS_OK);
+ bb.putLong(seq);
+ bb.putShort((short) 1); // tableCount
+ bb.putShort((short) name.length);
+ bb.put(name);
+ bb.putLong(seq);
+ return bb.array();
+ }
+ }
+
+ /**
+ * Connects a real client requesting the LOCAL tier and, like the production connect walk, refuses a
+ * server that did not grant it.
+ */
+ private static final class LocalTierWireFactory implements CursorWebSocketSendLoop.ReconnectFactory {
+ private final int port;
+
+ LocalTierWireFactory(int port) {
+ this.port = port;
+ }
+
+ @Override
+ public WebSocketClient reconnect() throws Exception {
+ WebSocketClient c = WebSocketClientFactory.newPlainTextInstance();
+ try {
+ c.setQwpMaxVersion(1);
+ c.setQwpDurableAckTiers(DurableAckTiers.LOCAL);
+ c.setConnectTimeout(5_000);
+ c.connect("localhost", port);
+ c.upgrade("/write/v4", 5_000, null);
+ if (!c.isServerDurableAckEnabled()) {
+ throw new QwpDurableAckMismatchException(
+ "localhost", port, DurableAckTiers.LOCAL, c.getServerDurableAckEcho());
+ }
+ } catch (Throwable t) {
+ c.close();
+ throw t;
+ }
+ return c;
+ }
+ }
+}
diff --git a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/BackgroundDrainerMidDrainAuthRejectTest.java b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/BackgroundDrainerMidDrainAuthRejectTest.java
index db5efa91..e8c221cf 100644
--- a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/BackgroundDrainerMidDrainAuthRejectTest.java
+++ b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/BackgroundDrainerMidDrainAuthRejectTest.java
@@ -24,6 +24,7 @@
package io.questdb.client.test.cutlass.qwp.client.sf.cursor;
+import io.questdb.client.cutlass.qwp.client.DurableAckTiers;
import io.questdb.client.SenderError;
import io.questdb.client.cutlass.http.client.WebSocketClient;
import io.questdb.client.cutlass.http.client.WebSocketClientFactory;
@@ -317,7 +318,7 @@ private BackgroundDrainer newDrainer(ScriptedWireFactory factory, long reconnect
reconnectMaxDurationMillis,
FAST_BACKOFF_MILLIS,
FAST_BACKOFF_MAX_MILLIS,
- /* requestDurableAck */ true,
+ /* durableAckTiers */ DurableAckTiers.REPLICATED,
/* durableAckKeepaliveIntervalMillis */ 200L);
}
@@ -559,7 +560,7 @@ public WebSocketClient reconnect() throws Exception {
WebSocketClient c = WebSocketClientFactory.newPlainTextInstance();
try {
c.setQwpMaxVersion(1);
- c.setQwpRequestDurableAck(true);
+ c.setQwpDurableAckTiers(DurableAckTiers.REPLICATED | DurableAckTiers.LEGACY_TRUE);
c.setConnectTimeout(5_000);
c.connect("localhost", port);
c.upgrade("/write/v4", 5_000, null);
diff --git a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/BackgroundDrainerMidDrainCapabilityGapTest.java b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/BackgroundDrainerMidDrainCapabilityGapTest.java
index 6f2224a1..4881ce23 100644
--- a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/BackgroundDrainerMidDrainCapabilityGapTest.java
+++ b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/BackgroundDrainerMidDrainCapabilityGapTest.java
@@ -24,6 +24,7 @@
package io.questdb.client.test.cutlass.qwp.client.sf.cursor;
+import io.questdb.client.cutlass.qwp.client.DurableAckTiers;
import io.questdb.client.cutlass.http.client.WebSocketClient;
import io.questdb.client.cutlass.http.client.WebSocketClientFactory;
import io.questdb.client.cutlass.http.client.WebSocketUpgradeException;
@@ -393,7 +394,7 @@ private BackgroundDrainer newDrainer(ScriptedWireFactory factory) {
RECONNECT_MAX_DURATION_MILLIS,
FAST_BACKOFF_MILLIS,
FAST_BACKOFF_MAX_MILLIS,
- /* requestDurableAck */ true,
+ /* durableAckTiers */ DurableAckTiers.REPLICATED,
/* durableAckKeepaliveIntervalMillis */ 200L);
}
@@ -645,7 +646,7 @@ public WebSocketClient reconnect() throws Exception {
WebSocketClient c = WebSocketClientFactory.newPlainTextInstance();
try {
c.setQwpMaxVersion(1);
- c.setQwpRequestDurableAck(true);
+ c.setQwpDurableAckTiers(DurableAckTiers.REPLICATED | DurableAckTiers.LEGACY_TRUE);
c.setConnectTimeout(5_000);
c.connect("localhost", port);
c.upgrade("/write/v4", 5_000, null);
diff --git a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/BackgroundDrainerOrphanTailTest.java b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/BackgroundDrainerOrphanTailTest.java
index 99f59817..895c489e 100644
--- a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/BackgroundDrainerOrphanTailTest.java
+++ b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/BackgroundDrainerOrphanTailTest.java
@@ -24,6 +24,7 @@
package io.questdb.client.test.cutlass.qwp.client.sf.cursor;
+import io.questdb.client.cutlass.qwp.client.DurableAckTiers;
import io.questdb.client.cutlass.qwp.client.sf.cursor.BackgroundDrainer;
import io.questdb.client.cutlass.qwp.client.sf.cursor.CursorSendEngine;
import io.questdb.client.cutlass.qwp.client.sf.cursor.OrphanScanner;
@@ -74,7 +75,7 @@ public void testDeferredOnlyRecoveredTailRetiresWithoutConnecting() throws Excep
5_000L,
1L,
5L,
- true,
+ DurableAckTiers.REPLICATED,
200L);
drainer.run();
@@ -132,7 +133,7 @@ public void testPreAdoptionSetupFailureDoesNotQuarantineTheSlot() throws Excepti
5_000L,
1L,
5L,
- true,
+ DurableAckTiers.REPLICATED,
200L);
drainer.run();
diff --git a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/BackgroundDrainerPoolConnectPhaseCloseTest.java b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/BackgroundDrainerPoolConnectPhaseCloseTest.java
index d31f60d4..3f2a5e12 100644
--- a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/BackgroundDrainerPoolConnectPhaseCloseTest.java
+++ b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/BackgroundDrainerPoolConnectPhaseCloseTest.java
@@ -24,6 +24,7 @@
package io.questdb.client.test.cutlass.qwp.client.sf.cursor;
+import io.questdb.client.cutlass.qwp.client.DurableAckTiers;
import io.questdb.client.cutlass.line.LineSenderException;
import io.questdb.client.cutlass.qwp.client.sf.cursor.BackgroundDrainer;
import io.questdb.client.cutlass.qwp.client.sf.cursor.BackgroundDrainerPool;
@@ -115,7 +116,7 @@ public void testCloseStopsConnectPhaseDrainerWithoutBurningGracefulWindow() thro
/* reconnectMaxDurationMillis */ 60_000L,
/* reconnectInitialBackoffMillis */ LONG_BACKOFF_MILLIS,
/* reconnectMaxBackoffMillis */ LONG_BACKOFF_MILLIS,
- /* requestDurableAck */ false,
+ /* durableAckTiers */ DurableAckTiers.NONE,
/* durableAckKeepaliveIntervalMillis */ 0L);
final BackgroundDrainerPool pool = new BackgroundDrainerPool(1);
pool.submit(drainer);
diff --git a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/BackgroundDrainerPoolInterruptedCloseTest.java b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/BackgroundDrainerPoolInterruptedCloseTest.java
index d8a3e82e..c6cecd02 100644
--- a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/BackgroundDrainerPoolInterruptedCloseTest.java
+++ b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/BackgroundDrainerPoolInterruptedCloseTest.java
@@ -24,6 +24,7 @@
package io.questdb.client.test.cutlass.qwp.client.sf.cursor;
+import io.questdb.client.cutlass.qwp.client.DurableAckTiers;
import io.questdb.client.cutlass.qwp.client.sf.cursor.BackgroundDrainer;
import io.questdb.client.cutlass.qwp.client.sf.cursor.BackgroundDrainerPool;
import io.questdb.client.cutlass.qwp.client.sf.cursor.CursorSendEngine;
@@ -130,7 +131,7 @@ public void interruptedCloseMustStopActivelyDrainingDrainers() throws Exception
firstAttempt.countDown();
throw new IOException("connection refused (test)");
},
- 5_000L, 1L, 10L, false, 0L);
+ 5_000L, 1L, 10L, DurableAckTiers.NONE, 0L);
BackgroundDrainerPool pool = new BackgroundDrainerPool(1);
try {
diff --git a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/BackgroundDrainerSetupFailureTest.java b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/BackgroundDrainerSetupFailureTest.java
index a8bd96c9..ceaf80b2 100644
--- a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/BackgroundDrainerSetupFailureTest.java
+++ b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/BackgroundDrainerSetupFailureTest.java
@@ -24,6 +24,7 @@
package io.questdb.client.test.cutlass.qwp.client.sf.cursor;
+import io.questdb.client.cutlass.qwp.client.DurableAckTiers;
import io.questdb.client.cutlass.qwp.client.sf.cursor.AckWatermark;
import io.questdb.client.cutlass.qwp.client.sf.cursor.BackgroundDrainer;
import io.questdb.client.cutlass.qwp.client.sf.cursor.CursorSendEngine;
@@ -75,7 +76,7 @@ public void testConnectErrorPropagatesWithoutQuarantine() throws Exception {
5_000L,
1L,
10L,
- true,
+ DurableAckTiers.REPLICATED,
200L);
LinkageError thrown = null;
@@ -123,7 +124,7 @@ public void testConstructionErrorPropagatesWithoutQuarantine() throws Exception
5_000L,
1L,
10L,
- true,
+ DurableAckTiers.REPLICATED,
200L);
OutOfMemoryError thrown = null;
@@ -182,7 +183,7 @@ public void testCorruptRecoveredChainIsQuarantined() throws Exception {
5_000L,
1L,
10L,
- true,
+ DurableAckTiers.REPLICATED,
200L);
drainer.run();
@@ -268,7 +269,7 @@ public void testSealedResidueFirstSightHealsAndDoesNotQuarantine() throws Except
5_000L,
1L,
10L,
- true,
+ DurableAckTiers.REPLICATED,
200L);
LinkageError thrown = null;
@@ -330,7 +331,7 @@ public void testLockOpenFailureDoesNotQuarantineRecoverableData() throws Excepti
5_000L,
1L,
10L,
- true,
+ DurableAckTiers.REPLICATED,
200L);
drainer.run();
@@ -370,7 +371,7 @@ public void testWatermarkOpenFailureDoesNotQuarantineRecoverableData() throws Ex
5_000L,
1L,
10L,
- true,
+ DurableAckTiers.REPLICATED,
200L);
drainer.run();
diff --git a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/BackgroundDrainerTransportOutageRecoveryTest.java b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/BackgroundDrainerTransportOutageRecoveryTest.java
index 86a3decc..f3278321 100644
--- a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/BackgroundDrainerTransportOutageRecoveryTest.java
+++ b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/BackgroundDrainerTransportOutageRecoveryTest.java
@@ -24,6 +24,7 @@
package io.questdb.client.test.cutlass.qwp.client.sf.cursor;
+import io.questdb.client.cutlass.qwp.client.DurableAckTiers;
import io.questdb.client.cutlass.http.client.WebSocketClient;
import io.questdb.client.cutlass.http.client.WebSocketClientFactory;
import io.questdb.client.cutlass.qwp.client.sf.cursor.BackgroundDrainer;
@@ -108,7 +109,7 @@ public void testDrainerSurvivesOutageLongerThanBudgetThenDrainsWhenServerReturns
RECONNECT_MAX_DURATION_MILLIS,
FAST_BACKOFF_MILLIS,
FAST_BACKOFF_MAX_MILLIS,
- /* requestDurableAck */ true,
+ /* durableAckTiers */ DurableAckTiers.REPLICATED,
/* durableAckKeepaliveIntervalMillis */ 200L);
CountingListener listener = new CountingListener();
drainer.setListener(listener);
@@ -305,7 +306,7 @@ public WebSocketClient reconnect() throws Exception {
WebSocketClient c = WebSocketClientFactory.newPlainTextInstance();
try {
c.setQwpMaxVersion(1);
- c.setQwpRequestDurableAck(true);
+ c.setQwpDurableAckTiers(DurableAckTiers.REPLICATED | DurableAckTiers.LEGACY_TRUE);
c.setConnectTimeout(5_000);
c.connect("localhost", port);
c.upgrade("/write/v4", 5_000, null);
diff --git a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/BackgroundDrainerUnreplayableSlotQuarantineTest.java b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/BackgroundDrainerUnreplayableSlotQuarantineTest.java
index a67a805d..c7a56af8 100644
--- a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/BackgroundDrainerUnreplayableSlotQuarantineTest.java
+++ b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/BackgroundDrainerUnreplayableSlotQuarantineTest.java
@@ -24,6 +24,7 @@
package io.questdb.client.test.cutlass.qwp.client.sf.cursor;
+import io.questdb.client.cutlass.qwp.client.DurableAckTiers;
import io.questdb.client.SenderError;
import io.questdb.client.cutlass.qwp.client.sf.cursor.BackgroundDrainer;
import io.questdb.client.cutlass.qwp.client.sf.cursor.MmapSegment;
@@ -92,7 +93,7 @@ public void testSecondDrainerNeverReAdoptsAQuarantinedSlot() throws Exception {
throw new AssertionError(
"recovery failure must be caught before any connect attempt");
},
- 5_000L, 1L, 5L, true, 200L);
+ 5_000L, 1L, 5L, DurableAckTiers.REPLICATED, 200L);
List
+ *
+ * {@link #LEGACY_TRUE} is a modifier bit for requests made through the
+ * boolean {@code requestDurableAck(true)} API or the {@code on} config
+ * value: the request header carries the literal {@code "true"} and the
+ * server confirms the grant with the {@code "enabled"} token. It always
+ * combines with {@link #REPLICATED}.
+ *
- *
+ * The server grants all of the requested tiers or none of them: it confirms a
+ * grant by echoing the requested token set in {@code X-QWP-Durable-Ack}
+ * ({@code enabled} for a legacy {@code on} request). A missing header, or an
+ * echo other than the expected token, is a denial. The message names the
+ * requested set and what the server echoed, or that it sent no header.
+ * {@link #getRole()} is {@code null} for this denial.
*/
public final class QwpDurableAckMismatchException extends HttpClientException {
private final String host;
private final int port;
private final String role;
+ /**
+ * The upgrade completed but the server did not grant the requested tier
+ * set. {@code serverEcho} is the server's {@code X-QWP-Durable-Ack} header
+ * value, or {@code null} when the server sent no such header.
+ */
+ public QwpDurableAckMismatchException(String host, int port, int requestedTiers, @Nullable String serverEcho) {
+ super("WebSocket upgrade failed: server does not support durable ack [host=");
+ put(host).put(", port=").put(port)
+ .put(", requested=").put(DurableAckTiers.configValue(requestedTiers))
+ .put(", echoed=");
+ if (serverEcho != null) {
+ put(serverEcho);
+ } else {
+ put("
* +--------+------------+--------------------------------------+
* | 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 0889d1ba..2e236811 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 7f04fa73..21d171ed 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;
@@ -294,6 +295,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
@@ -301,6 +307,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.
@@ -336,6 +347,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();
// Sender-lifetime observability counters: acks, frames sent/replayed,
// reconnect attempts and successes, server errors. A fresh instance per
@@ -628,6 +640,115 @@ public final class CursorWebSocketSendLoop implements QuietCloseable {
// only suppresses errors the user already owned before close() began.
private volatile LineSenderException synchronouslySurfacedError;
+ public CursorWebSocketSendLoop(WebSocketClient client, CursorSendEngine engine,
+ long fsnAtZero, long parkNanos,
+ ReconnectFactory reconnectFactory,
+ long reconnectInitialBackoffMillis,
+ long reconnectMaxBackoffMillis,
+ boolean durableAckMode) {
+ this(client, engine, fsnAtZero, parkNanos, reconnectFactory,
+ reconnectInitialBackoffMillis, reconnectMaxBackoffMillis,
+ durableAckMode ? DurableAckTiers.REPLICATED : DurableAckTiers.NONE);
+ }
+
+ public CursorWebSocketSendLoop(WebSocketClient client, CursorSendEngine engine,
+ long fsnAtZero, long parkNanos,
+ ReconnectFactory reconnectFactory,
+ long reconnectInitialBackoffMillis,
+ long reconnectMaxBackoffMillis,
+ boolean durableAckMode,
+ long durableAckKeepaliveIntervalMillis) {
+ this(client, engine, fsnAtZero, parkNanos, reconnectFactory,
+ reconnectInitialBackoffMillis, reconnectMaxBackoffMillis,
+ durableAckMode ? DurableAckTiers.REPLICATED : DurableAckTiers.NONE,
+ durableAckKeepaliveIntervalMillis);
+ }
+
+ public CursorWebSocketSendLoop(WebSocketClient client, CursorSendEngine engine,
+ long fsnAtZero, long parkNanos,
+ ReconnectFactory reconnectFactory,
+ long reconnectInitialBackoffMillis,
+ long reconnectMaxBackoffMillis,
+ boolean durableAckMode,
+ long durableAckKeepaliveIntervalMillis,
+ int maxHeadFrameRejections) {
+ this(client, engine, fsnAtZero, parkNanos, reconnectFactory,
+ reconnectInitialBackoffMillis, reconnectMaxBackoffMillis,
+ durableAckMode ? DurableAckTiers.REPLICATED : DurableAckTiers.NONE,
+ durableAckKeepaliveIntervalMillis, maxHeadFrameRejections);
+ }
+
+ public CursorWebSocketSendLoop(WebSocketClient client, CursorSendEngine engine,
+ long fsnAtZero, long parkNanos,
+ ReconnectFactory reconnectFactory,
+ long reconnectInitialBackoffMillis,
+ long reconnectMaxBackoffMillis,
+ boolean durableAckMode,
+ long durableAckKeepaliveIntervalMillis,
+ int maxHeadFrameRejections,
+ long poisonMinEscalationWindowMillis) {
+ this(client, engine, fsnAtZero, parkNanos, reconnectFactory,
+ reconnectInitialBackoffMillis, reconnectMaxBackoffMillis,
+ durableAckMode ? DurableAckTiers.REPLICATED : DurableAckTiers.NONE,
+ durableAckKeepaliveIntervalMillis, maxHeadFrameRejections,
+ poisonMinEscalationWindowMillis);
+ }
+
+ public CursorWebSocketSendLoop(WebSocketClient client, CursorSendEngine engine,
+ long fsnAtZero, long parkNanos,
+ ReconnectFactory reconnectFactory,
+ long reconnectInitialBackoffMillis,
+ long reconnectMaxBackoffMillis,
+ boolean durableAckMode,
+ long durableAckKeepaliveIntervalMillis,
+ int maxHeadFrameRejections,
+ long poisonMinEscalationWindowMillis,
+ long catchUpCapGapMinEscalationWindowMillis) {
+ this(client, engine, fsnAtZero, parkNanos, reconnectFactory,
+ reconnectInitialBackoffMillis, reconnectMaxBackoffMillis,
+ durableAckMode ? DurableAckTiers.REPLICATED : DurableAckTiers.NONE,
+ durableAckKeepaliveIntervalMillis, maxHeadFrameRejections,
+ poisonMinEscalationWindowMillis, catchUpCapGapMinEscalationWindowMillis);
+ }
+
+ public CursorWebSocketSendLoop(WebSocketClient client, CursorSendEngine engine,
+ long fsnAtZero, long parkNanos,
+ ReconnectFactory reconnectFactory,
+ long reconnectInitialBackoffMillis,
+ long reconnectMaxBackoffMillis,
+ boolean durableAckMode,
+ long durableAckKeepaliveIntervalMillis,
+ int maxHeadFrameRejections,
+ long poisonMinEscalationWindowMillis,
+ long catchUpCapGapMinEscalationWindowMillis,
+ ReconnectPolicy reconnectPolicy) {
+ this(client, engine, fsnAtZero, parkNanos, reconnectFactory,
+ reconnectInitialBackoffMillis, reconnectMaxBackoffMillis,
+ durableAckMode ? DurableAckTiers.REPLICATED : DurableAckTiers.NONE,
+ durableAckKeepaliveIntervalMillis, maxHeadFrameRejections,
+ poisonMinEscalationWindowMillis, catchUpCapGapMinEscalationWindowMillis,
+ reconnectPolicy, 0L);
+ }
+
+ public CursorWebSocketSendLoop(WebSocketClient client, CursorSendEngine engine,
+ long fsnAtZero, long parkNanos,
+ ReconnectFactory reconnectFactory,
+ long reconnectInitialBackoffMillis,
+ long reconnectMaxBackoffMillis,
+ int durableAckTiers,
+ long durableAckKeepaliveIntervalMillis,
+ int maxHeadFrameRejections,
+ long poisonMinEscalationWindowMillis,
+ long catchUpCapGapMinEscalationWindowMillis,
+ ReconnectPolicy reconnectPolicy,
+ long externalFsnBase) {
+ this(client, engine, fsnAtZero, parkNanos, reconnectFactory,
+ reconnectInitialBackoffMillis, reconnectMaxBackoffMillis, durableAckTiers,
+ durableAckKeepaliveIntervalMillis, maxHeadFrameRejections,
+ poisonMinEscalationWindowMillis, catchUpCapGapMinEscalationWindowMillis,
+ catchUpPolicyFor(reconnectPolicy), externalFsnBase);
+ }
+
/**
* Full constructor with explicit reconnect-policy knobs. When
* {@code reconnectFactory} is non-null, the I/O thread treats wire
@@ -652,13 +773,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
@@ -669,10 +791,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);
}
@@ -688,11 +810,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);
}
@@ -710,12 +832,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);
}
@@ -730,13 +852,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);
}
@@ -763,14 +885,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, 0L);
@@ -786,7 +908,7 @@ private CursorWebSocketSendLoop(WebSocketClient client, CursorSendEngine engine,
ReconnectFactory reconnectFactory,
long reconnectInitialBackoffMillis,
long reconnectMaxBackoffMillis,
- boolean durableAckMode,
+ int durableAckTiers,
long durableAckKeepaliveIntervalMillis,
int maxHeadFrameRejections,
long poisonMinEscalationWindowMillis,
@@ -949,7 +1071,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
@@ -982,17 +1105,14 @@ private CursorWebSocketSendLoop(WebSocketClient client, CursorSendEngine engine,
}
/**
- * The policy-aware constructor as released in 1.3.x: equivalent to the overload
- * below with {@code externalFsnBase = 0L}, the value every caller outside a
- * symbol-dictionary recycle passes. Kept so callers compiled against a released
- * jar keep linking.
+ * Tier-aware policy constructor with no external FSN offset.
*/
public CursorWebSocketSendLoop(WebSocketClient client, CursorSendEngine engine,
long fsnAtZero, long parkNanos,
ReconnectFactory reconnectFactory,
long reconnectInitialBackoffMillis,
long reconnectMaxBackoffMillis,
- boolean durableAckMode,
+ int durableAckTiers,
long durableAckKeepaliveIntervalMillis,
int maxHeadFrameRejections,
long poisonMinEscalationWindowMillis,
@@ -1000,7 +1120,7 @@ public CursorWebSocketSendLoop(WebSocketClient client, CursorSendEngine engine,
ReconnectPolicy reconnectPolicy) {
this(client, engine, fsnAtZero, parkNanos, reconnectFactory,
reconnectInitialBackoffMillis,
- reconnectMaxBackoffMillis, durableAckMode,
+ reconnectMaxBackoffMillis, durableAckTiers,
durableAckKeepaliveIntervalMillis, maxHeadFrameRejections,
poisonMinEscalationWindowMillis, catchUpCapGapMinEscalationWindowMillis,
reconnectPolicy, 0L);
@@ -1031,10 +1151,11 @@ public CursorWebSocketSendLoop(WebSocketClient client, CursorSendEngine engine,
long externalFsnBase) {
this(client, engine, fsnAtZero, parkNanos, reconnectFactory,
reconnectInitialBackoffMillis,
- reconnectMaxBackoffMillis, durableAckMode,
+ reconnectMaxBackoffMillis,
+ durableAckMode ? DurableAckTiers.REPLICATED : DurableAckTiers.NONE,
durableAckKeepaliveIntervalMillis, maxHeadFrameRejections,
poisonMinEscalationWindowMillis, catchUpCapGapMinEscalationWindowMillis,
- catchUpPolicyFor(reconnectPolicy), externalFsnBase);
+ reconnectPolicy, externalFsnBase);
}
private static CatchUpCapGapPolicy catchUpPolicyFor(ReconnectPolicy reconnectPolicy) {
@@ -1607,6 +1728,28 @@ 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();
+ }
+
+ /**
+ * The highest local-fsync seqTxn reported for the table via
+ * {@code STATUS_LOCAL_DURABLE_ACK}, or -1 when none arrived. Meaningful
+ * when both tiers are requested: the local frontier runs ahead of the
+ * replicated trim, and this exposes that early progress per table. In
+ * local-only mode the local acks feed the trim watermarks directly and
+ * this map stays empty.
+ */
+ public long getLocalDurableTableWatermark(CharSequence tableName) {
+ return localDurableTableWatermarks.get(tableName);
+ }
+
/**
* Total times a durable-ack frame caused {@link CursorSendEngine#acknowledge}
* to advance. Always 0 when {@code durableAckMode} is false. A non-zero
@@ -1898,6 +2041,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}).
@@ -1946,6 +2106,7 @@ private void clearDurableAckTracking() {
releasePendingEntry(pendingDurable.pollFirst());
}
durableTableWatermarks.clear();
+ localDurableTableWatermarks.clear();
}
private static boolean isCatchUpCapGap(Throwable t) {
@@ -4239,6 +4400,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 66ba0749..fbc833d8 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/DurableAckTiersTest.java b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/DurableAckTiersTest.java
new file mode 100644
index 00000000..87b65565
--- /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/MmapFaultDegradesTest.java b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/MmapFaultDegradesTest.java
index 360cbf5e..6e0bf773 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 4d463fa7..7ebdc5bc 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;
@@ -184,7 +185,7 @@ public void testErrorAtSuccessTailEntryClosesConnectedClient() throws Exception
// tail -- narrowing the try block later trips one of the two.
QwpWebSocketSender sender = newBareSender();
QwpHostHealthTracker tracker = wireEndpoints(sender, 1);
- setField(sender, "requestDurableAck", true);
+ setField(sender, "durableAckTiers", DurableAckTiers.REPLICATED | DurableAckTiers.LEGACY_TRUE);
OutOfMemoryError oom = new OutOfMemoryError("simulated allocation failure");
StubClient stub = newStubClient();
stub.durableAckCheckError = oom;
@@ -365,7 +366,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 3911f5c0..d90fd478 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 e305210b..1d1aaebf 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