Skip to content
44 changes: 23 additions & 21 deletions core/src/main/java/io/questdb/client/Sender.java
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -1452,17 +1452,17 @@ 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.
long defaultMaxTotal = sfDir == null
? 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
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -1638,7 +1638,7 @@ public Sender build() {
connected.startOrphanDrainers(
orphans,
maxBackgroundDrainers,
actualSfMaxBytes,
actualSfMaxSegmentBytes,
actualSfMaxTotalBytes,
actualSfSyncIntervalNanos);
}
Expand Down Expand Up @@ -2770,19 +2770,21 @@ public LineSenderBuilder storeAndForwardDurability(SfDurability durability) {
}

/**
* Maximum bytes per segment file before rotation. Defaults to
* {@code DEFAULT_SEGMENT_BYTES}
* (4 MiB). Smaller segments mean faster trim of acked data; larger
* segments mean fewer rotations.
* Maximum bytes per segment file before rotation, the builder form of
* the {@code sf_max_segment_bytes} connect-string key. Smaller segments
* mean faster trim of acked data; larger segments mean fewer rotations.
* Default: {@code 4 MiB}. WebSocket transport only.
*
* @param maxSegmentBytes per-segment cap in bytes; must be positive
*/
public LineSenderBuilder storeAndForwardMaxBytes(long maxBytes) {
public LineSenderBuilder storeAndForwardMaxSegmentBytes(long maxSegmentBytes) {
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);
if (maxSegmentBytes <= 0) {
throw new LineSenderException("sf_max_segment_bytes must be positive: ").put(maxSegmentBytes);
}
this.sfMaxBytes = maxBytes;
this.sfMaxSegmentBytes = maxSegmentBytes;
return this;
}

Expand Down Expand Up @@ -3387,12 +3389,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");
Expand Down Expand Up @@ -3730,8 +3732,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"));
Expand Down Expand Up @@ -3892,7 +3894,7 @@ public java.util.Map<String, Object> 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);
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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<PendingDurableEntry> pendingDurable = new ArrayDeque<>();
private final ArrayDeque<PendingDurableEntry> pendingDurablePool = new ArrayDeque<>();
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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);
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -38,7 +38,7 @@
/**
* Tests for WebSocket transport support in the Sender.builder() API.
* These tests verify the builder configuration and validation,
* not actual WebSocket connectivity (which requires a running server).
* not actual WebSocket connectivity (that requires a running server).
*/
public class LineSenderBuilderWebSocketTest extends AbstractTest {

Expand Down Expand Up @@ -634,6 +634,39 @@ 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) {
assertThrows("sf_max_segment_bytes must be positive: " + value,
() -> Sender.builder(Sender.Transport.WEBSOCKET).storeAndForwardMaxSegmentBytes(value));
}
}

@Test
public void testStoreAndForwardMaxSegmentBytesRejectedForNonWebSocketTransports() {
Sender.Transport[] rejected = {
Sender.Transport.HTTP,
Sender.Transport.TCP,
Sender.Transport.UDP
};
for (Sender.Transport transport : rejected) {
assertThrows("store_and_forward is only supported for WebSocket transport",
() -> Sender.builder(transport).storeAndForwardMaxSegmentBytes(64 * 1024L));
}
}

@Test
public void testSyncModeAutoFlushDefaults() throws Exception {
// Regression test: connect() must not hardcode autoFlush to 0.
Expand Down Expand Up @@ -809,6 +842,15 @@ public void testWsConfigString_withPath_fails() {
"unknown configuration key: path");
}

@Test
public void testWsConfigString_withSfMaxBytes_fails() {
// sf_max_bytes is the spelling this key carries in the 1.2.1-1.3.5 jars.
// sf_max_segment_bytes replaces it with no alias, so the old spelling
// rejects as an unknown key rather than silently configuring nothing.
assertBadConfig("ws::addr=localhost:9000;sf_max_bytes=4096;",
"unknown configuration key: sf_max_bytes");
}

@Test
public void testWsConfigString_withEgressOnlyKeysSilentlyAccepted() {
// connect-string.md "Query client keys" and "Multi-host failover": these
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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",
};
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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++) {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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)) {
Expand All @@ -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++) {
Expand Down Expand Up @@ -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.
Expand Down Expand Up @@ -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)) {
Expand All @@ -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();
Expand Down Expand Up @@ -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"));
}
});
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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.
* <p>
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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);
Expand Down
Loading