diff --git a/core/src/main/java/io/questdb/client/Sender.java b/core/src/main/java/io/questdb/client/Sender.java index e2891201..64ee70e4 100644 --- a/core/src/main/java/io/questdb/client/Sender.java +++ b/core/src/main/java/io/questdb/client/Sender.java @@ -1128,7 +1128,7 @@ public int getConnectTimeout() { // Durability contract for SF append/flush. FLUSH and APPEND remain // deferred follow-ups; PERIODIC uses the segment manager. private SfDurability sfDurability = SfDurability.MEMORY; - private long sfMaxBytes = PARAMETER_NOT_SET_EXPLICITLY; + private long sfMaxSegmentBytes = PARAMETER_NOT_SET_EXPLICITLY; private long sfMaxTotalBytes = PARAMETER_NOT_SET_EXPLICITLY; private long sfSyncIntervalMillis = PARAMETER_NOT_SET_EXPLICITLY; private boolean shouldDestroyPrivKey; @@ -1452,9 +1452,9 @@ public Sender build() { // (same lock-free architecture, no disk involvement). // Durability-combination validation lives in validateParameters // so build() and no-connect validation apply the same rules. - long actualSfMaxBytes = sfMaxBytes == PARAMETER_NOT_SET_EXPLICITLY + long actualSfMaxSegmentBytes = sfMaxSegmentBytes == PARAMETER_NOT_SET_EXPLICITLY ? DEFAULT_SEGMENT_BYTES - : sfMaxBytes; + : sfMaxSegmentBytes; // Default cap depends on backing: RAM (memory mode) is tight // by default; disk (SF mode) is cheap so the default is // generous enough that normal traffic never hits it. @@ -1462,7 +1462,7 @@ public Sender build() { ? DEFAULT_MAX_BYTES_MEMORY : DEFAULT_MAX_BYTES_SF; long actualSfMaxTotalBytes = sfMaxTotalBytes == PARAMETER_NOT_SET_EXPLICITLY - ? Math.max(defaultMaxTotal, actualSfMaxBytes * 2) + ? Math.max(defaultMaxTotal, actualSfMaxSegmentBytes * 2) : sfMaxTotalBytes; long actualCloseFlushTimeoutMillis = closeFlushTimeoutMillis == CLOSE_FLUSH_TIMEOUT_NOT_SET ? DEFAULT_CLOSE_FLUSH_TIMEOUT_MILLIS @@ -1556,7 +1556,7 @@ public Sender build() { ? DEFAULT_SF_SYNC_INTERVAL_MILLIS : sfSyncIntervalMillis) * 1_000_000L : 0L; CursorSendEngine cursorEngine = new CursorSendEngine( - slotPath, actualSfMaxBytes, + slotPath, actualSfMaxSegmentBytes, actualSfMaxTotalBytes, actualSfAppendDeadlineNanos, actualSfSyncIntervalNanos); int actualErrorInboxCapacity = errorInboxCapacity != PARAMETER_NOT_SET_EXPLICITLY @@ -1638,7 +1638,7 @@ public Sender build() { connected.startOrphanDrainers( orphans, maxBackgroundDrainers, - actualSfMaxBytes, + actualSfMaxSegmentBytes, actualSfMaxTotalBytes, actualSfSyncIntervalNanos); } @@ -2775,14 +2775,14 @@ public LineSenderBuilder storeAndForwardDurability(SfDurability durability) { * (4 MiB). Smaller segments mean faster trim of acked data; larger * segments mean fewer rotations. */ - public LineSenderBuilder storeAndForwardMaxBytes(long maxBytes) { + public LineSenderBuilder storeAndForwardMaxSegmentBytes(long maxBytes) { if (protocol != PARAMETER_NOT_SET_EXPLICITLY && protocol != PROTOCOL_WEBSOCKET) { throw new LineSenderException("store_and_forward is only supported for WebSocket transport"); } if (maxBytes <= 0) { - throw new LineSenderException("sf_max_bytes must be positive: ").put(maxBytes); + throw new LineSenderException("sf_max_segment_bytes must be positive: ").put(maxBytes); } - this.sfMaxBytes = maxBytes; + this.sfMaxSegmentBytes = maxBytes; return this; } @@ -3387,12 +3387,12 @@ private LineSenderBuilder fromConfig(CharSequence configurationString) { } pos = getValue(configurationString, pos, sink, "sender_id"); senderId(sink.toString()); - } else if (Chars.equals("sf_max_bytes", sink)) { + } else if (Chars.equals("sf_max_segment_bytes", sink)) { if (protocol != PROTOCOL_WEBSOCKET) { - throw new LineSenderException("sf_max_bytes is only supported for WebSocket transport"); + throw new LineSenderException("sf_max_segment_bytes is only supported for WebSocket transport"); } - pos = getValue(configurationString, pos, sink, "sf_max_bytes"); - storeAndForwardMaxBytes(parseSizeValue(sink, "sf_max_bytes")); + pos = getValue(configurationString, pos, sink, "sf_max_segment_bytes"); + storeAndForwardMaxSegmentBytes(parseSizeValue(sink, "sf_max_segment_bytes")); } else if (Chars.equals("sf_max_total_bytes", sink)) { if (protocol != PROTOCOL_WEBSOCKET) { throw new LineSenderException("sf_max_total_bytes is only supported for WebSocket transport"); @@ -3730,8 +3730,8 @@ private LineSenderBuilder fromConfigWebSocket(CharSequence configurationString) if (view.has("sf_append_deadline_millis")) { sfAppendDeadlineMillis(wsLong(view, v, "sf_append_deadline_millis")); } - if (view.has("sf_max_bytes")) { - storeAndForwardMaxBytes(wsSize(view, v, "sf_max_bytes")); + if (view.has("sf_max_segment_bytes")) { + storeAndForwardMaxSegmentBytes(wsSize(view, v, "sf_max_segment_bytes")); } if (view.has("sf_max_total_bytes")) { storeAndForwardMaxTotalBytes(wsSize(view, v, "sf_max_total_bytes")); @@ -3892,7 +3892,7 @@ public java.util.Map wsConfigSnapshotForTest() { m.put("request_durable_ack", requestDurableAck); m.put("sender_id", senderId); m.put("sf_dir", sfDir); - m.put("sf_max_bytes", sfMaxBytes); + m.put("sf_max_segment_bytes", sfMaxSegmentBytes); m.put("sf_max_total_bytes", sfMaxTotalBytes); m.put("sf_durability", sfDurability == null ? null : sfDurability.name()); m.put("sf_append_deadline_millis", sfAppendDeadlineMillis); 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 58c3d31e..3f58fa35 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 @@ -183,7 +183,7 @@ public final class CursorWebSocketSendLoop implements QuietCloseable { // drained from the head every time a STATUS_DURABLE_ACK frame advances // any watermark; an entry pops when every (name, seqTxn) it carries is // covered by durableTableWatermarks. Bounded in practice by the SF on-disk - // cap: once the producer hits sf_max_bytes it blocks, which caps how far + // cap: once the producer hits sf_max_segment_bytes it blocks, which caps how far // the durable watermark can lag behind the OK watermark. private final ArrayDeque pendingDurable = new ArrayDeque<>(); private final ArrayDeque pendingDurablePool = new ArrayDeque<>(); diff --git a/core/src/main/java/io/questdb/client/cutlass/qwp/client/sf/cursor/MmapSegment.java b/core/src/main/java/io/questdb/client/cutlass/qwp/client/sf/cursor/MmapSegment.java index edf70252..d7d2b44d 100644 --- a/core/src/main/java/io/questdb/client/cutlass/qwp/client/sf/cursor/MmapSegment.java +++ b/core/src/main/java/io/questdb/client/cutlass/qwp/client/sf/cursor/MmapSegment.java @@ -219,7 +219,7 @@ static MmapSegment create(FilesFacade ff, long pathPtr, String displayPath, long // the JVM). if (!ff.allocate(fd, sizeBytes)) { ff.close(fd); - // Unlink the partially-created file so a sf_max_bytes-sized + // Unlink the partially-created file so a sf_max_segment_bytes-sized // empty file does not survive the failure. Under sustained // disk-full pressure with the manager polling, hundreds would // otherwise accumulate. diff --git a/core/src/main/java/io/questdb/client/impl/ConfigSchema.java b/core/src/main/java/io/questdb/client/impl/ConfigSchema.java index 9cd18134..de8cce43 100644 --- a/core/src/main/java/io/questdb/client/impl/ConfigSchema.java +++ b/core/src/main/java/io/questdb/client/impl/ConfigSchema.java @@ -86,7 +86,7 @@ public final class ConfigSchema { str("sf_append_deadline_millis", Side.INGRESS); str("sf_dir", Side.INGRESS); str("sf_durability", Side.INGRESS); - str("sf_max_bytes", Side.INGRESS); + str("sf_max_segment_bytes", Side.INGRESS); str("sf_max_total_bytes", Side.INGRESS); str("sf_sync_interval_millis", Side.INGRESS); str("transaction", Side.INGRESS); diff --git a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/LineSenderBuilderWebSocketTest.java b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/LineSenderBuilderWebSocketTest.java index caec264b..9ec82093 100644 --- a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/LineSenderBuilderWebSocketTest.java +++ b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/LineSenderBuilderWebSocketTest.java @@ -634,6 +634,54 @@ public void testRetryTimeout_notSupportedForWebSocket() { "not supported for WebSocket"); } + @Test + public void testStoreAndForwardMaxSegmentBytes() { + Sender.LineSenderBuilder builder = Sender.builder(Sender.Transport.WEBSOCKET); + + Assert.assertSame(builder, builder.storeAndForwardMaxSegmentBytes(64 * 1024L)); + Assert.assertEquals( + 64 * 1024L, + ((Number) builder.wsConfigSnapshotForTest().get("sf_max_segment_bytes")).longValue() + ); + } + + @Test + public void testStoreAndForwardMaxSegmentBytesRejectsNonPositiveValues() { + long[] rejected = {0L, -1L}; + for (long value : rejected) { + try { + Sender.builder(Sender.Transport.WEBSOCKET) + .storeAndForwardMaxSegmentBytes(value); + Assert.fail("expected storeAndForwardMaxSegmentBytes(" + value + ") to fail"); + } catch (LineSenderException expected) { + Assert.assertEquals( + "sf_max_segment_bytes must be positive: " + value, + expected.getMessage() + ); + } + } + } + + @Test + public void testStoreAndForwardMaxSegmentBytesRejectedForNonWebSocketTransports() { + Sender.Transport[] rejected = { + Sender.Transport.HTTP, + Sender.Transport.TCP, + Sender.Transport.UDP + }; + for (Sender.Transport transport : rejected) { + try { + Sender.builder(transport).storeAndForwardMaxSegmentBytes(64 * 1024L); + Assert.fail("expected " + transport + " to reject storeAndForwardMaxSegmentBytes"); + } catch (LineSenderException expected) { + Assert.assertEquals( + "store_and_forward is only supported for WebSocket transport", + expected.getMessage() + ); + } + } + } + @Test public void testSyncModeAutoFlushDefaults() throws Exception { // Regression test: connect() must not hardcode autoFlush to 0. diff --git a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/QwpQueryClientFromConfigTest.java b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/QwpQueryClientFromConfigTest.java index 45f34fef..42d668fb 100644 --- a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/QwpQueryClientFromConfigTest.java +++ b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/QwpQueryClientFromConfigTest.java @@ -720,7 +720,7 @@ public void testIngressOnlyKeysSilentlyAcceptedOnEgress() { "sf_append_deadline_millis=30000", "sf_dir=/var/lib/qdb-sf", "sf_durability=memory", - "sf_max_bytes=4m", + "sf_max_segment_bytes=4m", "sf_max_total_bytes=10g", "transaction=on", }; diff --git a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/RecoveryReplayTest.java b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/RecoveryReplayTest.java index e868dafc..98a59e6f 100644 --- a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/RecoveryReplayTest.java +++ b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/RecoveryReplayTest.java @@ -86,7 +86,7 @@ public void testRestartReplaysSealedSegmentsAgainstFreshServer() throws Exceptio String pad = repeat("x", 64); String cfg1 = "ws::addr=localhost:" + port1 + ";sf_dir=" + sfDir - + ";sf_max_bytes=4096" + + ";sf_max_segment_bytes=4096" + ";close_flush_timeout_millis=0;"; try (Sender s1 = Sender.fromConfig(cfg1)) { for (int i = 0; i < 50; i++) { diff --git a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/SfFromConfigTest.java b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/SfFromConfigTest.java index fc2f1833..78be527c 100644 --- a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/SfFromConfigTest.java +++ b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/SfFromConfigTest.java @@ -93,7 +93,7 @@ public void testSfDirOnTcpRejected() throws Exception { } @Test - public void testSfMaxBytesParsing() throws Exception { + public void testSfMaxSegmentBytesParsing() throws Exception { TestUtils.assertMemoryLeak(() -> { AckHandler handler = new AckHandler(); try (TestWebSocketServer server = new TestWebSocketServer(handler)) { @@ -102,7 +102,7 @@ public void testSfMaxBytesParsing() throws Exception { int port = server.getPort(); String config = "ws::addr=localhost:" + port - + ";sf_dir=" + sfDir + ";sf_max_bytes=131072;"; + + ";sf_dir=" + sfDir + ";sf_max_segment_bytes=131072;"; try (Sender sender = Sender.fromConfig(config)) { // Write enough data that segments rotate at ~128 KiB boundary. for (int i = 0; i < 50; i++) { @@ -140,7 +140,7 @@ public void testNoSfDirMeansNoSf() throws Exception { } /** - * Regression test for the connect-string {@code sf_max_bytes} / + * Regression test for the connect-string {@code sf_max_segment_bytes} / * {@code sf_max_total_bytes} parser accepting values larger than * {@code Integer.MAX_VALUE}. The pre-cursor parser used parseInt which * artificially capped the SF size from the connect string at ~2 GiB. @@ -349,7 +349,7 @@ public void testSfSyncIntervalRejectsOnNonWebSocketTransport() throws Exception } @Test - public void testSfMaxBytesAcceptsSizeSuffixes() throws Exception { + public void testSfMaxSegmentBytesAcceptsSizeSuffixes() throws Exception { TestUtils.assertMemoryLeak(() -> { AckHandler handler = new AckHandler(); try (TestWebSocketServer server = new TestWebSocketServer(handler)) { @@ -360,7 +360,7 @@ public void testSfMaxBytesAcceptsSizeSuffixes() throws Exception { // 64m / 4g should parse identically to their byte-count equivalents. String config = "ws::addr=localhost:" + port + ";sf_dir=" + sfDir - + ";sf_max_bytes=64m" + + ";sf_max_segment_bytes=64m" + ";sf_max_total_bytes=4g;"; try (Sender sender = Sender.fromConfig(config)) { sender.table("foo").longColumn("v", 1L).atNow(); @@ -473,14 +473,14 @@ public void testSenderIdInvalidCharRejected() throws Exception { } @Test - public void testSfMaxBytesInvalidSizeSuffixRejected() throws Exception { + public void testSfMaxSegmentBytesInvalidSizeSuffixRejected() throws Exception { TestUtils.assertMemoryLeak(() -> { - String config = "ws::addr=localhost:1;sf_dir=" + sfDir + ";sf_max_bytes=64x;"; + String config = "ws::addr=localhost:1;sf_dir=" + sfDir + ";sf_max_segment_bytes=64x;"; try (Sender ignored = Sender.fromConfig(config)) { Assert.fail("expected rejection of unknown unit suffix"); } catch (LineSenderException expected) { Assert.assertTrue(expected.getMessage(), - expected.getMessage().contains("invalid sf_max_bytes")); + expected.getMessage().contains("invalid sf_max_segment_bytes")); } }); } diff --git a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/SegmentRingTest.java b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/SegmentRingTest.java index d000cd3a..988dbb85 100644 --- a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/SegmentRingTest.java +++ b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/SegmentRingTest.java @@ -1004,7 +1004,7 @@ public void testNextSealedAfterSurvivesConcurrentRotationAndHeadTrim() throws Ex /** * Open-time sort regression: at the documented {@code sf_max_total_bytes - * / sf_max_bytes} ceiling (~16K segments), an O(N²) sort over the + * / sf_max_segment_bytes} ceiling (~16K segments), an O(N²) sort over the * recovered segments delays the I/O thread. This test guards the sort * with a deterministic comparison count. *

diff --git a/core/src/test/java/io/questdb/client/test/impl/WsSenderConfigHonoredTest.java b/core/src/test/java/io/questdb/client/test/impl/WsSenderConfigHonoredTest.java index 7fe602d3..e7be38ed 100644 --- a/core/src/test/java/io/questdb/client/test/impl/WsSenderConfigHonoredTest.java +++ b/core/src/test/java/io/questdb/client/test/impl/WsSenderConfigHonoredTest.java @@ -61,7 +61,7 @@ public void testEveryIngressKeyIsHonored() { assertHonored("request_durable_ack=on", "request_durable_ack", true); assertHonored("sender_id=probe-1", "sender_id", "probe-1"); assertHonored("sf_dir=/var/probe", "sf_dir", "/var/probe"); - assertHonored("sf_max_bytes=4096", "sf_max_bytes", 4096L); + assertHonored("sf_max_segment_bytes=4096", "sf_max_segment_bytes", 4096L); assertHonored("sf_max_total_bytes=8192", "sf_max_total_bytes", 8192L); assertHonored("sf_durability=periodic", "sf_durability", "PERIODIC"); assertHonored("sf_append_deadline_millis=1500", "sf_append_deadline_millis", 1500L);