diff --git a/README.md b/README.md index 2fa3c75..c108177 100644 --- a/README.md +++ b/README.md @@ -312,7 +312,7 @@ The config string format is: | `request_durable_ack` | `off` | `on` / `off` — opts into per-table object-store ACK watermarks (cast to `IQwpWebSocketSender`). | | `sf_dir` | | Path to a local directory enabling store-and-forward. Sets the SF stack on this sender. | | `sender_id` | `default` | Slot identifier under `//`. Must be unique per process sharing the same `sf_dir`. | -| `sf_max_bytes` | `4194304` | Per-segment rotation threshold in bytes (default 4 MiB). | +| `sf_max_segment_bytes` | `4194304` | Per-segment rotation threshold in bytes (default 4 MiB). | | `sf_max_total_bytes` | `10 GiB` with `sf_dir`, `128 MiB` otherwise | Hard cap on total disk usage; back-pressures the producer when exceeded. | | `sf_durability` | `memory` | Durability mode. Only `memory` is supported in v1. | | `sf_append_deadline_millis` | `30000` | Max wait when the disk cap is hit before `Send` throws. | diff --git a/src/net-questdb-client-tests/QuestDbDockerIngestFaultToleranceTests.cs b/src/net-questdb-client-tests/QuestDbDockerIngestFaultToleranceTests.cs index 504f0a6..9bc9e4c 100644 --- a/src/net-questdb-client-tests/QuestDbDockerIngestFaultToleranceTests.cs +++ b/src/net-questdb-client-tests/QuestDbDockerIngestFaultToleranceTests.cs @@ -301,10 +301,10 @@ public async Task SchemaMismatch_HaltsLoudly_NotSilentlyDropped() public async Task SfExhaustion_SurfacesAsAppendBackpressure_NotTerminal() { var table = NewTable(); - // sf_max_total_bytes must be >= 2 * sf_max_bytes (room for a hot spare). Pooled Send is a + // sf_max_total_bytes must be >= 2 * sf_max_segment_bytes (room for a hot spare). Pooled Send is a // non-draining flush-to-ring, so the tiny store fills and surfaces append backpressure. await using var client = QuestDBClient.Connect(WsConn( - "sf_dir=" + NewSfDir() + ";sf_max_bytes=524288;sf_max_total_bytes=1048576;" + + "sf_dir=" + NewSfDir() + ";sf_max_segment_bytes=524288;sf_max_total_bytes=1048576;" + "sf_append_deadline_millis=2000;sender_pool_min=1;query_pool_min=0;")); await PauseAsync(); diff --git a/src/net-questdb-client-tests/Qwp/QwpWebSocketSenderTests.cs b/src/net-questdb-client-tests/Qwp/QwpWebSocketSenderTests.cs index db29a51..da59373 100644 --- a/src/net-questdb-client-tests/Qwp/QwpWebSocketSenderTests.cs +++ b/src/net-questdb-client-tests/Qwp/QwpWebSocketSenderTests.cs @@ -962,7 +962,7 @@ public async Task EndToEnd_Sf_AutoFlushByRows_RoutesThroughEngine_NotTransport() { using var sender = NewSender(server, $"auto_flush_rows=2;auto_flush_interval=off;auto_flush_bytes=off;" + - $"sf_dir={sfRoot};sender_id=af-test;sf_max_bytes=4096;"); + $"sf_dir={sfRoot};sender_id=af-test;sf_max_segment_bytes=4096;"); for (var i = 0; i < 5; i++) { @@ -989,7 +989,7 @@ public async Task EndToEnd_Sf_EveryFrame_IsSelfSufficient_AcrossMultipleFlushes( try { using (var sender = NewSender(server, - $"auto_flush=off;sf_dir={sfRoot};sender_id=svc-multi;sf_max_bytes=4096;")) + $"auto_flush=off;sf_dir={sfRoot};sender_id=svc-multi;sf_max_segment_bytes=4096;")) { for (var i = 0; i < 3; i++) { @@ -1025,7 +1025,7 @@ public async Task EndToEnd_Sf_SingleRow_FrameReachesServerAndIsSelfSufficient() try { using (var sender = NewSender(server, - $"auto_flush=off;sf_dir={sfRoot};sender_id=svc-a;sf_max_bytes=4096;")) + $"auto_flush=off;sf_dir={sfRoot};sender_id=svc-a;sf_max_segment_bytes=4096;")) { sender.Table("trades") .Symbol("ticker", "ETH-USD") @@ -1114,7 +1114,7 @@ public async Task Sf_Flush_PersistsRowsToDisk_DisposeDiscardsUnsent() { using (var sender = NewSender(server, $"auto_flush=on;auto_flush_rows=1;auto_flush_interval=off;auto_flush_bytes=off;" + - $"sf_dir={sentRoot};sender_id=svc-a;sf_max_bytes=4096;")) + $"sf_dir={sentRoot};sender_id=svc-a;sf_max_segment_bytes=4096;")) { sender.Table("trades").Symbol("ticker", "ETH-USD").Column("price", 2615.54) .At(new DateTime(2026, 4, 28, 12, 0, 0, DateTimeKind.Utc)); // auto-flush → on-disk ring @@ -1133,7 +1133,7 @@ public async Task Sf_Flush_PersistsRowsToDisk_DisposeDiscardsUnsent() try { using (var sender = NewSender(server, - $"auto_flush=off;sf_dir={unsentRoot};sender_id=svc-b;sf_max_bytes=4096;")) + $"auto_flush=off;sf_dir={unsentRoot};sender_id=svc-b;sf_max_segment_bytes=4096;")) { sender.Table("trades").Symbol("ticker", "ETH-USD").Column("price", 2615.54) .At(new DateTime(2026, 4, 28, 12, 0, 0, DateTimeKind.Utc)); diff --git a/src/net-questdb-client-tests/SenderOptionsTests.cs b/src/net-questdb-client-tests/SenderOptionsTests.cs index 2f6c402..dd8e15c 100644 --- a/src/net-questdb-client-tests/SenderOptionsTests.cs +++ b/src/net-questdb-client-tests/SenderOptionsTests.cs @@ -217,7 +217,7 @@ public void Sf_DefaultsAreSane() var opts = new SenderOptions("ws::addr=localhost:9000;"); Assert.That(opts.sf_dir, Is.Null); Assert.That(opts.sender_id, Is.EqualTo("default")); - Assert.That(opts.sf_max_bytes, Is.EqualTo(4L * 1024 * 1024)); + Assert.That(opts.sf_max_segment_bytes, Is.EqualTo(4L * 1024 * 1024)); Assert.That(opts.sf_max_total_bytes, Is.EqualTo(128L * 1024 * 1024)); Assert.That(opts.sf_durability, Is.EqualTo("memory")); Assert.That(opts.sf_append_deadline_millis, Is.EqualTo(TimeSpan.FromSeconds(30))); @@ -242,7 +242,7 @@ public void Sf_AllKeysParse() { var opts = new SenderOptions( "wss::addr=questdb.io:9000;sf_dir=/var/qdb-sf;sender_id=svc-7;" + - "sf_max_bytes=1048576;sf_max_total_bytes=10485760;sf_durability=memory;" + + "sf_max_segment_bytes=1048576;sf_max_total_bytes=10485760;sf_durability=memory;" + "sf_append_deadline_millis=10000;reconnect_max_duration_millis=60000;" + "reconnect_initial_backoff_millis=200;reconnect_max_backoff_millis=5000;" + "initial_connect_retry=on;close_flush_timeout_millis=2000;" + @@ -250,7 +250,7 @@ public void Sf_AllKeysParse() Assert.That(opts.sf_dir, Is.EqualTo("/var/qdb-sf")); Assert.That(opts.sender_id, Is.EqualTo("svc-7")); - Assert.That(opts.sf_max_bytes, Is.EqualTo(1048576L)); + Assert.That(opts.sf_max_segment_bytes, Is.EqualTo(1048576L)); Assert.That(opts.sf_max_total_bytes, Is.EqualTo(10485760L)); Assert.That(opts.sf_durability, Is.EqualTo("memory")); Assert.That(opts.sf_append_deadline_millis, Is.EqualTo(TimeSpan.FromMilliseconds(10000))); @@ -1196,7 +1196,7 @@ public void Sf_AllKeysOnHttpScheme_RejectedIndividually() { var keys = new[] { - "sender_id=foo", "sf_max_bytes=1024", "sf_max_total_bytes=1024", "sf_durability=memory", + "sender_id=foo", "sf_max_segment_bytes=1024", "sf_max_total_bytes=1024", "sf_durability=memory", "sf_append_deadline_millis=1000", "reconnect_max_duration_millis=1000", "reconnect_initial_backoff_millis=1", "reconnect_max_backoff_millis=1", "initial_connect_retry=on", "close_flush_timeout_millis=100", @@ -1425,19 +1425,19 @@ public void Proxy_OnHttpScheme_String_Rejected() } [Test] - public void SfMaxTotalBytes_LessThanTwiceSfMaxBytes_Rejected() + public void SfMaxTotalBytes_LessThanTwiceSfMaxSegmentBytes_Rejected() { Assert.That( - () => new SenderOptions("ws::addr=h:9000;sf_dir=/tmp/test;sf_max_bytes=8000000;sf_max_total_bytes=10000000;"), + () => new SenderOptions("ws::addr=h:9000;sf_dir=/tmp/test;sf_max_segment_bytes=8000000;sf_max_total_bytes=10000000;"), Throws.TypeOf().With.Message.Contains("sf_max_total_bytes")); } [Test] - public void SfMaxBytes_NonPositive_Rejected() + public void SfMaxSegmentBytes_NonPositive_Rejected() { Assert.That( - () => new SenderOptions("ws::addr=h:9000;sf_dir=/tmp/test;sf_max_bytes=0;"), - Throws.TypeOf().With.Message.Contains("sf_max_bytes")); + () => new SenderOptions("ws::addr=h:9000;sf_dir=/tmp/test;sf_max_segment_bytes=0;"), + Throws.TypeOf().With.Message.Contains("sf_max_segment_bytes")); } [Test] diff --git a/src/net-questdb-client/Qwp/QwpConnectStringKeys.cs b/src/net-questdb-client/Qwp/QwpConnectStringKeys.cs index e90ad67..34f7f5a 100644 --- a/src/net-questdb-client/Qwp/QwpConnectStringKeys.cs +++ b/src/net-questdb-client/Qwp/QwpConnectStringKeys.cs @@ -64,7 +64,7 @@ internal static class QwpConnectStringKeys "auto_flush", "auto_flush_rows", "auto_flush_bytes", "auto_flush_interval", "init_buf_size", "max_buf_size", "max_name_len", "convert_local_to_utc", "request_durable_ack", - "sf_dir", "sender_id", "sf_max_bytes", "sf_max_total_bytes", "sf_durability", + "sf_dir", "sender_id", "sf_max_segment_bytes", "sf_max_total_bytes", "sf_durability", "sf_append_deadline_millis", "reconnect_max_duration_millis", "reconnect_initial_backoff_millis", "reconnect_max_backoff_millis", "initial_connect_retry", "close_flush_timeout_millis", diff --git a/src/net-questdb-client/Qwp/Sf/QwpSegmentRing.cs b/src/net-questdb-client/Qwp/Sf/QwpSegmentRing.cs index 4d5d2a3..636b9b1 100644 --- a/src/net-questdb-client/Qwp/Sf/QwpSegmentRing.cs +++ b/src/net-questdb-client/Qwp/Sf/QwpSegmentRing.cs @@ -325,7 +325,7 @@ public bool TryAppend(ReadOnlySpan frame) if (frame.Length + QwpMmapSegment.EnvelopeHeaderSize > _segmentCapacity) { throw new ArgumentException( - $"frame ({frame.Length} bytes) exceeds segment capacity ({_segmentCapacity} bytes); raise sf_max_bytes", + $"frame ({frame.Length} bytes) exceeds segment capacity ({_segmentCapacity} bytes); raise sf_max_segment_bytes", nameof(frame)); } diff --git a/src/net-questdb-client/Senders/QwpWebSocketSender.cs b/src/net-questdb-client/Senders/QwpWebSocketSender.cs index 345d1a1..3dc60c6 100644 --- a/src/net-questdb-client/Senders/QwpWebSocketSender.cs +++ b/src/net-questdb-client/Senders/QwpWebSocketSender.cs @@ -144,7 +144,7 @@ private static (QwpSlotLock? slotLock, QwpCursorSendEngine engine, QwpBackground var sfRoot = options.sf_dir!; var slotDir = Path.Combine(sfRoot, options.sender_id); slotLock = QwpSlotLock.Acquire(slotDir); - ring = QwpSegmentRing.Open(slotDir, segmentCapacity: options.sf_max_bytes); + ring = QwpSegmentRing.Open(slotDir, segmentCapacity: options.sf_max_segment_bytes); if (ring.NextFsn == 0) { // Clear any stale watermark from a prior session that left no segments behind. @@ -154,7 +154,7 @@ private static (QwpSlotLock? slotLock, QwpCursorSendEngine engine, QwpBackground } else { - ring = QwpSegmentRing.OpenMemoryBacked(segmentCapacity: options.sf_max_bytes); + ring = QwpSegmentRing.OpenMemoryBacked(segmentCapacity: options.sf_max_segment_bytes); } var authHeader = BuildAuthHeader(options); @@ -225,7 +225,7 @@ private static (QwpSlotLock? slotLock, QwpCursorSendEngine engine, QwpBackground () => !drainerTracker.IsRoundExhausted); }, policy, - segmentCapacity: options.sf_max_bytes, + segmentCapacity: options.sf_max_segment_bytes, drainTimeout: options.reconnect_max_duration_millis, durableAckMode: options.request_durable_ack, maxFrameRejections: options.max_frame_rejections, diff --git a/src/net-questdb-client/Utils/SenderOptions.cs b/src/net-questdb-client/Utils/SenderOptions.cs index 27449a6..647b4ac 100644 --- a/src/net-questdb-client/Utils/SenderOptions.cs +++ b/src/net-questdb-client/Utils/SenderOptions.cs @@ -94,7 +94,7 @@ public record SenderOptions private string? _sfDir; private string _senderId = "default"; - private long _sfMaxBytes = 4L * 1024 * 1024; + private long _sfMaxSegmentBytes = 4L * 1024 * 1024; private long _sfMaxTotalBytes = 128L * 1024 * 1024; private string _sfDurability = "memory"; private TimeSpan _sfAppendDeadline = TimeSpan.FromMilliseconds(30000); @@ -143,7 +143,7 @@ public record SenderOptions private bool _transactionUserSet; private bool _sfDirUserSet; private bool _senderIdUserSet; - private bool _sfMaxBytesUserSet; + private bool _sfMaxSegmentBytesUserSet; private bool _sfMaxTotalBytesUserSet; private bool _sfDurabilityUserSet; private bool _sfAppendDeadlineUserSet; @@ -284,8 +284,8 @@ public SenderOptions(string confStr) ParseStringWithDefault(nameof(sf_dir), null, out _sfDir); ParseStringWithDefault(nameof(sender_id), "default", out var senderIdRaw); SetSenderId(senderIdRaw ?? "default"); - ParseLongWithDefault(nameof(sf_max_bytes), - (4L * 1024 * 1024).ToString(System.Globalization.CultureInfo.InvariantCulture), out _sfMaxBytes); + ParseLongWithDefault(nameof(sf_max_segment_bytes), + (4L * 1024 * 1024).ToString(System.Globalization.CultureInfo.InvariantCulture), out _sfMaxSegmentBytes); _sfMaxTotalBytesUserSet = ReadOptionFromBuilder(nameof(sf_max_total_bytes)) is not null; var defaultMaxTotal = string.IsNullOrEmpty(_sfDir) ? 128L * 1024 * 1024 @@ -829,30 +829,30 @@ private void ValidateStoreAndForwardOptions() _sfMaxTotalBytes = 10L * 1024 * 1024 * 1024; } - if (_sfMaxBytes <= 0) + if (_sfMaxSegmentBytes <= 0) { throw new IngressError(ErrorCode.ConfigError, - $"`sf_max_bytes` must be > 0; got {_sfMaxBytes}"); + $"`sf_max_segment_bytes` must be > 0; got {_sfMaxSegmentBytes}"); } // A segment is buffered as a single contiguous byte[] (QwpCursorSendEngine's send buffer), // so it must fit a managed array; reject early as a clean ConfigError instead of letting an // oversized value surface as an opaque OverflowException/OOM at engine construction. - if (_sfMaxBytes > Array.MaxLength) + if (_sfMaxSegmentBytes > Array.MaxLength) { throw new IngressError(ErrorCode.ConfigError, - $"`sf_max_bytes` ({_sfMaxBytes}) exceeds the maximum single-segment size of {Array.MaxLength} bytes"); + $"`sf_max_segment_bytes` ({_sfMaxSegmentBytes}) exceeds the maximum single-segment size of {Array.MaxLength} bytes"); } if (_sfMaxTotalBytes <= 0) { throw new IngressError(ErrorCode.ConfigError, $"`sf_max_total_bytes` must be > 0; got {_sfMaxTotalBytes}"); } - // Overflow-safe form of `_sfMaxTotalBytes < 2 * _sfMaxBytes`; the doubled multiply - // wraps negative for huge sf_max_bytes and would defeat the check. - if (_sfMaxTotalBytes / 2 < _sfMaxBytes) + // Overflow-safe form of `_sfMaxTotalBytes < 2 * _sfMaxSegmentBytes`; the doubled multiply + // wraps negative for huge sf_max_segment_bytes and would defeat the check. + if (_sfMaxTotalBytes / 2 < _sfMaxSegmentBytes) { throw new IngressError(ErrorCode.ConfigError, - $"`sf_max_total_bytes` ({_sfMaxTotalBytes}) must be >= 2 * `sf_max_bytes` ({_sfMaxBytes}) so the segment manager has room to provision a hot spare."); + $"`sf_max_total_bytes` ({_sfMaxTotalBytes}) must be >= 2 * `sf_max_segment_bytes` ({_sfMaxSegmentBytes}) so the segment manager has room to provision a hot spare."); } } @@ -867,7 +867,7 @@ private void ValidateWebSocketKeysAgainstDefaults() if (_transactionUserSet) Throw(nameof(transaction)); if (_sfDirUserSet) Throw(nameof(sf_dir)); if (_senderIdUserSet) Throw(nameof(sender_id)); - if (_sfMaxBytesUserSet) Throw(nameof(sf_max_bytes)); + if (_sfMaxSegmentBytesUserSet) Throw(nameof(sf_max_segment_bytes)); if (_sfMaxTotalBytesUserSet) Throw(nameof(sf_max_total_bytes)); if (_sfDurabilityUserSet) Throw(nameof(sf_durability)); if (_sfAppendDeadlineUserSet) Throw(nameof(sf_append_deadline_millis)); @@ -925,7 +925,7 @@ private static string[] BuildWebSocketOnlyKeys() var keys = new List { "request_durable_ack", "transaction", - "sf_dir", "sender_id", "sf_max_bytes", "sf_max_total_bytes", "sf_durability", + "sf_dir", "sender_id", "sf_max_segment_bytes", "sf_max_total_bytes", "sf_durability", "sf_append_deadline_millis", "reconnect_max_duration_millis", "reconnect_initial_backoff_millis", "reconnect_max_backoff_millis", "initial_connect_retry", "initial_connect_mode", "close_flush_timeout_millis", "drain_orphans", "max_background_drainers", "ping_timeout", @@ -1549,10 +1549,10 @@ private void SetSenderId(string value) } /// Per-segment rotation threshold in bytes. Defaults to 4 MiB. - public long sf_max_bytes + public long sf_max_segment_bytes { - get => _sfMaxBytes; - set { _sfMaxBytes = value; _sfMaxBytesUserSet = true; } + get => _sfMaxSegmentBytes; + set { _sfMaxSegmentBytes = value; _sfMaxSegmentBytesUserSet = true; } } ///