Skip to content

refactor(qwp)!: rename sf_max_bytes to sf_max_segment_bytes #79

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
mtopolnik wants to merge 1 commit into
base: main
Choose a base branch
from
Open

refactor(qwp)!: rename sf_max_bytes to sf_max_segment_bytes #79

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 `<sf_dir>/<sender_id>/`. 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. |
Expand Down
4 changes: 2 additions & 2 deletions src/net-questdb-client-tests/QuestDbDockerIngestFaultToleranceTests.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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();
Expand Down
10 changes: 5 additions & 5 deletions src/net-questdb-client-tests/Qwp/QwpWebSocketSenderTests.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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++)
{
Expand All @@ -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++)
{
Expand Down Expand Up @@ -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")
Expand Down Expand Up @@ -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
Expand All @@ -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));
Expand Down
18 changes: 9 additions & 9 deletions src/net-questdb-client-tests/SenderOptionsTests.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)));
Expand All @@ -242,15 +242,15 @@ 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;" +
"drain_orphans=on;max_background_drainers=8;");

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)));
Expand Down Expand Up @@ -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",
Expand Down Expand Up @@ -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<IngressError>().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<IngressError>().With.Message.Contains("sf_max_bytes"));
() => new SenderOptions("ws::addr=h:9000;sf_dir=/tmp/test;sf_max_segment_bytes=0;"),
Throws.TypeOf<IngressError>().With.Message.Contains("sf_max_segment_bytes"));
}

[Test]
Expand Down
2 changes: 1 addition & 1 deletion src/net-questdb-client/Qwp/QwpConnectStringKeys.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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",
Expand Down
2 changes: 1 addition & 1 deletion src/net-questdb-client/Qwp/Sf/QwpSegmentRing.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -325,7 +325,7 @@ public bool TryAppend(ReadOnlySpan<byte> 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));
}

Expand Down
6 changes: 3 additions & 3 deletions src/net-questdb-client/Senders/QwpWebSocketSender.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand All @@ -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);
Expand Down Expand Up @@ -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,
Expand Down
34 changes: 17 additions & 17 deletions src/net-questdb-client/Utils/SenderOptions.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);
Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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.");
}
}

Expand All @@ -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));
Expand Down Expand Up @@ -925,7 +925,7 @@ private static string[] BuildWebSocketOnlyKeys()
var keys = new List<string>
{
"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",
Expand Down Expand Up @@ -1549,10 +1549,10 @@ private void SetSenderId(string value)
}

/// <summary>Per-segment rotation threshold in bytes. Defaults to 4 MiB.</summary>
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; }
}

/// <summary>
Expand Down
Loading