feat(qwp): WebSocket ingest sender + egress query client + store-and-forward#73
Open
feat(qwp): WebSocket ingest sender + egress query client + store-and-forward#73
Conversation
|
Note Reviews pausedIt looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the Use the following commands to manage reviews:
Use the checkboxes below for quick actions:
📝 WalkthroughWalkthroughAdds WebSocket (ws/wss) QWP ingestion with QWP v1 encoder/decoder, Gorilla timestamp compression, a ClientWebSocket transport, store-and-forward (SF) persistence and cursor engine, extensive tests/benchmarks/examples, README/docs updates, and small public API extensions (ISender async dispose, new IQwpWebSocketSender, protocol/type/status enums). Changes
Sequence Diagram(s)sequenceDiagram
participant App as Application
participant Sender as QwpWebSocketSender
participant Transport as QwpWebSocketTransport
participant Server as QuestDB (QWP)
participant SF as QwpCursorSendEngine
rect rgba(200,200,255,0.5)
App->>Sender: Append rows / Send / SendAsync
Sender->>Sender: Encode frame (QwpEncoder)
alt sf_dir configured
Sender->>SF: Persist frame to mmap segment ring
SF->>Sender: Persist complete
SF->>Transport: Replay SendBinaryAsync(frame)
else direct WS
Sender->>Transport: SendBinaryAsync(frame)
end
Transport->>Server: WebSocket binary QWP frame
Server-->>Transport: OK / Durable-ACK
Transport->>Sender: Deliver parsed QwpResponse
Sender->>Sender: Update in-flight window & durable watermarks
Sender-->>App: Send/SendAsync completes
end
Estimated code review effort🎯 5 (Critical) | ⏱️ ~120 minutes Suggested reviewers
✨ Finishing Touches🧪 Generate unit tests (beta)
|
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 18
🧹 Nitpick comments (11)
src/example-websocket-auth-tls/Program.cs (1)
20-23: Avoid hardcoded credentials and insecure TLS defaults in runnable examples.Line 22 hardcodes
username/passwordandtls_verify=unsafe_off. Even with comments, this pattern is frequently copy-pasted into non-test code. Prefer environment-driven values so unsafe settings are explicit at runtime.🔐 Suggested safer example pattern
-using var sender = - Sender.New( - "wss::addr=localhost:9000;username=admin;password=quest;tls_verify=unsafe_off;request_durable_ack=on;"); +var username = Environment.GetEnvironmentVariable("QDB_USERNAME") ?? "admin"; +var password = Environment.GetEnvironmentVariable("QDB_PASSWORD") ?? "quest"; +var tlsVerify = Environment.GetEnvironmentVariable("QDB_TLS_VERIFY") ?? "on"; +using var sender = + Sender.New( + $"wss::addr=localhost:9000;username={username};password={password};tls_verify={tlsVerify};request_durable_ack=on;");🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/example-websocket-auth-tls/Program.cs` around lines 20 - 23, The example currently hardcodes credentials and disables TLS verification in the Sender.New connection string; replace that literal with environment-driven values and a safe default for TLS verification: read username, password, host/port and tls_verify from environment variables (e.g., via Environment.GetEnvironmentVariable) and fall back to secure defaults (tls_verify=on) if not provided, and ensure Sender.New is passed the constructed connection string rather than the hardcoded one so unsafe settings are explicit at runtime; update any comments to instruct users to set ENV vars for testing instead of copying credentials into source.src/net-questdb-client/Qwp/QwpInFlightWindow.cs (1)
196-237: Consider eliminating poll-based cancellation inAwaitEmpty.Line 237 currently depends on 100ms polling to observe cancellation, which adds avoidable wakeups and cancellation latency. You can register a cancellation callback that pulses the monitor and then wait indefinitely when no timeout is active.
♻️ Suggested refinement
public void AwaitEmpty(TimeSpan timeout, CancellationToken ct = default) { + using var ctr = ct.CanBeCanceled + ? ct.Register(static state => + { + var gate = (object)state!; + lock (gate) Monitor.PulseAll(gate); + }, _lock) + : default; + var hasDeadline = timeout >= TimeSpan.Zero; @@ - int waitMs; + int waitMs; if (hasDeadline) { @@ } else { - waitMs = CancellationPollMs; + waitMs = Timeout.Infinite; } Monitor.Wait(_lock, waitMs); } } }🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/net-questdb-client/Qwp/QwpInFlightWindow.cs` around lines 196 - 237, The AwaitEmpty method currently uses a poll loop with CancellationPollMs and Monitor.Wait(_lock, waitMs) to observe cancellation, causing extra wakeups and latency; instead, register ct.Register(() => { lock(_lock) Monitor.PulseAll(_lock); }) when ct.CanBeCanceled and, when no deadline is set (hasDeadline == false), call Monitor.Wait(_lock) (indefinite wait) so the cancellation callback will wake the waiter; ensure the registration is disposed/unregistered before returning or throwing and preserve existing timeout behavior when hasDeadline is true.src/net-questdb-client-tests/Qwp/QwpVarintTests.cs (2)
140-158: Consider simplifying random value generation.The current approach uses
(hi << 32) | lowherehiis already a 64-bit value, effectively discarding the upper 32 bits ofhi. For full 64-bit random coverage, you could simplify:Simplified random ulong generation
for (var i = 0; i < 10_000; i++) { - var hi = (ulong)rnd.NextInt64(); - var lo = (uint)rnd.Next(); - var value = (hi << 32) | lo; + var value = (ulong)rnd.NextInt64(); var written = QwpVarint.Write(buffer, value);Or if you need unsigned distribution across the full range:
Span<byte> bytes = stackalloc byte[8]; rnd.NextBytes(bytes); var value = BinaryPrimitives.ReadUInt64LittleEndian(bytes);🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/net-questdb-client-tests/Qwp/QwpVarintTests.cs` around lines 140 - 158, The test RoundTrip_RandomFuzz_PreservesValue currently builds a 64-bit value with (hi << 32) | lo which discards the upper 32 bits of hi; instead generate a full random ulong before writing/reading. Update the value generation in the test (inside RoundTrip_RandomFuzz_PreservesValue) to produce a true 64-bit unsigned value—e.g. fill an 8-byte stackalloc buffer with rnd.NextBytes and use BinaryPrimitives.ReadUInt64LittleEndian, or combine two 32-bit Next() calls correctly—to ensure QwpVarint.Write/ QwpVarint.Read are exercised across the full 64-bit range.
79-89: Minor: Redundant ternary condition.The condition
bit == 63 ? 1ul << 63 : 1ul << bitis unnecessary since1ul << bitworks correctly for all bit values 0-63 in C#.Simplify expression
for (var bit = 0; bit < 64; bit++) { - var value = bit == 63 ? 1ul << 63 : 1ul << bit; + var value = 1ul << bit; AssertRoundTrip(value); if (value > 0) AssertRoundTrip(value - 1); }🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/net-questdb-client-tests/Qwp/QwpVarintTests.cs` around lines 79 - 89, The test RoundTrip_PowerOfTwoBoundaries_PreservesValue uses a redundant ternary to set value (bit == 63 ? 1ul << 63 : 1ul << bit); simplify by replacing that expression with the single safe shift 1ul << bit and keep the surrounding logic (the loop over bit and calls to AssertRoundTrip and the conditional AssertRoundTrip(value - 1)) unchanged so behavior is identical but cleaner.src/net-questdb-client-benchmarks/BenchSfAppend.cs (1)
86-100: Empty catch blocks may mask cleanup failures.While swallowing exceptions during cleanup is common practice, consider logging at debug level to aid troubleshooting during benchmark development.
[GlobalCleanup] public async Task Cleanup() { - try { _sfSender?.Dispose(); } catch { } - try { _wsSender?.Dispose(); } catch { } + try { _sfSender?.Dispose(); } catch (Exception ex) { Console.WriteLine($"SF sender dispose: {ex.Message}"); } + try { _wsSender?.Dispose(); } catch (Exception ex) { Console.WriteLine($"WS sender dispose: {ex.Message}"); } if (_server is not null) { await _server.DisposeAsync(); } if (Directory.Exists(_sfRoot)) { - try { Directory.Delete(_sfRoot, recursive: true); } catch { } + try { Directory.Delete(_sfRoot, recursive: true); } catch (Exception ex) { Console.WriteLine($"SF root cleanup: {ex.Message}"); } } }🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/net-questdb-client-benchmarks/BenchSfAppend.cs` around lines 86 - 100, The Cleanup method currently swallows all exceptions (empty catch blocks) for _sfSender.Dispose(), _wsSender.Dispose(), _server.DisposeAsync(), and Directory.Delete(_sfRoot), which can hide cleanup failures; replace each empty catch with a catch (Exception ex) that logs the exception at debug level (e.g., _logger.Debug/LogDebug or Console.Error.WriteLine if no logger exists) including context (which resource failed) so failures in Dispose/DisposeAsync and Directory.Delete are recorded; ensure you capture exceptions for the asynchronous server dispose (await inside try/catch around _server.DisposeAsync()) and include the resource name (_sfSender, _wsSender, _server, _sfRoot) in each log message.src/net-questdb-client/Qwp/Sf/QwpBackgroundDrainerPool.cs (2)
146-156: SecondTask.WhenAllcall after cancellation is redundant.The same
snapshotarray is used withTask.WhenAlltwice. After cancellation, you could await the sameTaskobject returned from the firstTask.WhenAllcall instead of creating a new one.♻️ Reuse the WhenAll task
var allJoined = snapshot.Length == 0; + Task? joinTask = null; if (snapshot.Length > 0) { + joinTask = Task.WhenAll(snapshot); try { - allJoined = Task.WhenAll(snapshot).Wait(_shutdownWait); + allJoined = joinTask.Wait(_shutdownWait); } catch (Exception) { // Drain failures already wrote .failed sentinels; swallow here. } SfCleanup.Run(() => _shutdownCts.Cancel()); try { - allJoined = Task.WhenAll(snapshot).Wait(TimeSpan.FromSeconds(2)); + allJoined = joinTask.Wait(TimeSpan.FromSeconds(2)); }🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/net-questdb-client/Qwp/Sf/QwpBackgroundDrainerPool.cs` around lines 146 - 156, The code calls Task.WhenAll(snapshot) twice; instead capture the Task returned from the first Task.WhenAll into a local (e.g. whenAllTask) and reuse it after invoking SfCleanup.Run(() => _shutdownCts.Cancel()) instead of calling Task.WhenAll(snapshot) again; update the variable currently assigned to allJoined to Wait(TimeSpan.FromSeconds(2)) on that stored whenAllTask (and catch exceptions as before) so you don't allocate/duplicate the same aggregate Task for the same snapshot array.
100-106:ContinueWithuses default scheduler without continuation options.The continuation uses
TaskScheduler.Defaultbut doesn't specifyTaskContinuationOptions. If the antecedent task faults, the continuation still runs (which is correct here), but consider usingTaskContinuationOptions.ExecuteSynchronouslyto avoid a thread-pool hop for this lightweight cleanup.♻️ Optional: run continuation inline
_ = task.ContinueWith(t => { lock (_trackingLock) { _runningTasks.Remove(t); } - }, TaskScheduler.Default); + }, CancellationToken.None, TaskContinuationOptions.ExecuteSynchronously, TaskScheduler.Default);🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/net-questdb-client/Qwp/Sf/QwpBackgroundDrainerPool.cs` around lines 100 - 106, The continuation created in QwpBackgroundDrainerPool (the task.ContinueWith block that locks _trackingLock and calls _runningTasks.Remove(t)) should be executed synchronously to avoid an unnecessary thread-pool hop; modify the ContinueWith call to include TaskContinuationOptions.ExecuteSynchronously (keep the current TaskScheduler.Default or omit scheduler if not needed) so the lightweight cleanup runs inline when possible.src/net-questdb-client-tests/Qwp/QwpWebSocketSenderTests.cs (1)
556-562: Consider extracting SF cleanup to a helper method.The same cleanup pattern (
if (Directory.Exists(sfRoot)) try { Directory.Delete(sfRoot, recursive: true); } catch { }) is repeated across multiple SF tests. A small helper would reduce duplication.♻️ Optional cleanup helper
private static void TryDeleteDirectory(string path) { if (Directory.Exists(path)) { try { Directory.Delete(path, recursive: true); } catch { } } }Then replace each occurrence with
TryDeleteDirectory(sfRoot);🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/net-questdb-client-tests/Qwp/QwpWebSocketSenderTests.cs` around lines 556 - 562, Extract the repeated SF cleanup block into a single helper method (e.g., add a private static TryDeleteDirectory(string path) in the QwpWebSocketSenderTests class) that performs the current pattern (check Directory.Exists, then try Directory.Delete(path, recursive: true) with an empty catch), and replace each inline occurrence like the one using sfRoot with a call to TryDeleteDirectory(sfRoot); this reduces duplication and keeps behavior identical.src/net-questdb-client/Qwp/QwpColumn.cs (1)
737-749: Potential null dereference if called without prior allocation.
EnsureOffsetCapacityusesStrOffsets!assuming it's already allocated. WhileAppendVarchar(the only caller currently) initializesStrOffsetsbefore calling this method, a future caller could trigger aNullReferenceException.This is a minor concern since the class is
internaland the current usage is safe.🛡️ Suggested defensive check
private void EnsureOffsetCapacity(int requiredCount) { + if (StrOffsets is null) + { + StrOffsets = new uint[Math.Max(InitialSymbolCapacity, requiredCount)]; + StrOffsets[0] = 0; + return; + } + - if (StrOffsets!.Length < requiredCount) + if (StrOffsets.Length < requiredCount) {🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/net-questdb-client/Qwp/QwpColumn.cs` around lines 737 - 749, EnsureOffsetCapacity currently dereferences StrOffsets (StrOffsets!) and can NRE if called before allocation; modify EnsureOffsetCapacity to defensively handle null by initializing StrOffsets to a sensible starting array (e.g., length 1 or requiredCount rounded up to power-of-two) before running the resize loop, and then proceed with the existing doubling logic to grow to requiredCount; reference EnsureOffsetCapacity and the StrOffsets field (and note AppendVarchar currently initializes it) so the change is focused and keeps existing behavior for current callers.src/net-questdb-client-tests/Qwp/QwpEncoderTests.cs (1)
582-592: Consider removing unused parameters.The
userColCountandframeparameters are captured but immediately discarded (lines 586-587). While the comment mentions "kept for future multi-column encoder fixtures," unused parameters can cause confusion.🔧 Suggested cleanup
Either remove the unused parameters if they're not needed soon, or add a
// TODO: use for multi-column validationcomment to clarify intent:- private static int FindFirstColumnDataOffset(byte[] frame, int tableNameLen, int userColCount, int userColDefSize) + private static int FindFirstColumnDataOffset(int tableNameLen, int userColDefSize) { - // userColCount unused for the simple single-user-column tests above; kept for future - // multi-column encoder fixtures. - _ = userColCount; - _ = frame; - // 12 (header) + 2 (delta dict) + 1 (table name varint) + tableNameLen + 1 (rowCount) + 1 (colCount) // + 2 (schema mode + id) + userColDefSize + 2 (designated TS def: empty name varint=0 + TIMESTAMP) return 12 + 2 + 1 + tableNameLen + 1 + 1 + 2 + userColDefSize + 2; }🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/net-questdb-client-tests/Qwp/QwpEncoderTests.cs` around lines 582 - 592, The method FindFirstColumnDataOffset currently takes unused parameters frame and userColCount which are immediately discarded; either remove these parameters from the signature and all call sites (leaving tableNameLen and userColDefSize only), or keep them and replace the discard lines with an explicit clarifying comment like "// TODO: reserved for future multi-column encoder validation" and remove the "_ = frame;" since it's unused; update any references to userColCount/frame across tests to match the new signature if you choose removal, or leave them in place but document intent if you choose to retain them.src/net-questdb-client/Qwp/Sf/QwpMmapSegment.cs (1)
282-310: Consider suppressing finalizer if no native resources are held directly.The class implements
IDisposablebut doesn't callGC.SuppressFinalize(this). WhileMemoryMappedFileandMemoryMappedViewAccessorhandle their own finalization, if you're relying on deterministic disposal viausingor explicitDispose(), addingSuppressFinalizeis a minor optimization to skip the finalizer queue check.🔧 Suggested addition
public void Dispose() { if (_disposed) { return; } _disposed = true; + GC.SuppressFinalize(this); try { _view.Flush(); }🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/net-questdb-client/Qwp/Sf/QwpMmapSegment.cs` around lines 282 - 310, The Dispose method in QwpMmapSegment.cs doesn't call GC.SuppressFinalize(this); add a call to GC.SuppressFinalize(this) (e.g. at the end of Dispose after setting _disposed = true) to avoid unnecessary finalization overhead when consumers dispose the object; no changes to _view/_mmap cleanup logic are needed unless you add a finalizer, in which case ensure the finalizer and Dispose(bool) pattern are implemented correctly.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@docs/qwip-benchmarks.md`:
- Line 1: The file name `qwip-benchmarks.md` is inconsistent with the document
content and PR which use "QWP"; rename the file to `qwp-benchmarks.md` and
update any internal references or links (e.g., the top-level header "# .NET
WebSocket / QWP — Performance Benchmarks" and any README or docs index entries)
to use the new filename and the "QWP" spelling so naming is consistent across
the repo.
In `@net-questdb-client.sln`:
- Around line 176-178: The solution's NestedProjects section only maps the
example-websocket-auth-tls project to the src folder GUID
({A1FE95A9-4761-4806-8891-A82F468624F8} =
{827E0CD3-B72D-47B6-A68D-7590B98EB39B}); add the missing mapping for the
example-websocket project by inserting its project GUID mapped to the same src
folder GUID in the GlobalSection(NestedProjects) so both example-websocket and
example-websocket-auth-tls are nested under the src solution folder.
In `@README.md`:
- Around line 198-200: The README’s “User code never sees disconnects” claim
overpromises; update the Store-and-forward section to state that setting sf_dir
enables an on-disk buffer that masks transient disconnects but does not provide
an unconditional guarantee—explicitly reference the bounded retry/drain settings
(sf_append_deadline_millis and reconnect_max_duration_millis) and note that
prolonged outages can still cause Send to fail or the sender to become terminal;
adjust the language around sf_dir and the store-and-forward description to say
it “masks transient disconnects” and include a short note pointing readers to
sf_append_deadline_millis and reconnect_max_duration_millis for failure/timeout
behavior.
In `@src/net-questdb-client-benchmarks/BenchLatencyWs.cs`:
- Around line 86-88: The code uses a probe-then-bind pattern (GetFreeTcpPort()
then DummyHttpServer.StartAsync(httpPort)) which is racy; change to let the
server bind atomically by passing port 0 (or providing an overload that binds to
port 0) and then read the actual bound port from the server (e.g., expose an
AssignedPort/Port property or return the bound port from StartAsync) so you
remove GetFreeTcpPort() and avoid TOCTOU; update any callers (BenchLatencyWs
startup and the 133-140 block) to use StartAsync(0) or the new overload and then
read the assigned port from DummyHttpServer instead of relying on
GetFreeTcpPort().
In `@src/net-questdb-client-tests/QuestDbWebSocketIntegrationTests.cs`:
- Around line 146-147: The test DurableAck_OnRequestDurableAck_PopulatesSeqTxn
is asserting the wrong value by calling GetHighestAckedSeqTxn, which checks the
acked transaction but not the durable watermark; update the assertion to query
the durable watermark instead (replace the call to
ws.GetHighestAckedSeqTxn("test_ws_durable") with the appropriate
durable-watermark accessor, e.g. ws.GetDurableWatermark("test_ws_durable") or
similarly named method on the websocket client) and assert that returned durable
watermark is >= 0 to validate durable watermark population.
In `@src/net-questdb-client-tests/Qwp/QwpSchemaCacheTests.cs`:
- Around line 107-118: Add a test that after calling QwpSchemaCache.Reset() and
then calling PrepareSchema(t) on a previously-used QwpTableBuffer (with its
existing SchemaId) the returned mode equals SchemaModeFull; specifically, reuse
the existing test setup (create QwpSchemaCache, create QwpTableBuffer "t", call
PrepareSchema(t), call cache.Reset()), then call PrepareSchema(t) again and
Assert.That(returnedMode, Is.EqualTo(SchemaModeFull)) to ensure Reset causes the
cache to treat the table as new rather than SchemaModeReference.
In `@src/net-questdb-client-tests/Qwp/QwpTableBufferTests.cs`:
- Around line 179-183: The tests currently append two different-typed values to
the same column within the same row so they hit the duplicate-write guard
instead of the column-type check; update the tests (the ones using
QwpTableBuffer and methods AppendLong/AppendDouble, and the similar test at
lines 212-216) to perform the second append in a new row (e.g., call NewRow or
advance to the next row between appends) so the duplicate-write guard is avoided
and the QwpColumn type-mismatch path is exercised and asserted via
Assert.Throws<IngressError>.
In `@src/net-questdb-client-tests/Qwp/Sf/QwpSegmentManagerTests.cs`:
- Around line 137-149: The test races because mgr.SparesInstalled is sampled
after the appends but TryAllocateNewActive can wake and install the replacement
spare during those appends; move the sparesBefore sampling to immediately after
the initial WaitFor(() => mgr.SparesInstalled >= 1, ...) and before the
ring.TryAppend calls so you compare against the pre-append spare count (i.e.,
capture mgr.SparesInstalled right after WaitFor), keeping the rest of the
assertions (WaitFor(() => mgr.SparesInstalled > sparesBefore, ...), Stopwatch
and the HeartbeatInterval assertion) unchanged and referencing the same symbols
(WaitFor, mgr.SparesInstalled, ring.TryAppend, TryAllocateNewActive,
QwpSegmentManager.HeartbeatInterval).
In `@src/net-questdb-client-tests/SenderOptionsTests.cs`:
- Around line 317-325: The test MultiAddress_RejectedForWebSocket currently uses
duplicate keys and therefore checks last-writer-wins behavior rather than the
documented multi-address syntax; update the test so SenderOptions is constructed
with the documented single-key multi-host form (e.g., a single "addr" value
containing multiple hosts in the documented separator format) for both "ws:" and
"wss:" variants and assert it throws IngressError; ensure the test targets the
SenderOptions constructor (and any parsing helper it uses) so it will fail when
a true multi-address string is passed for WebSocket schemes.
In `@src/net-questdb-client/Qwp/Sf/QwpFiles.cs`:
- Around line 67-77: TryOpenExclusive currently swallows every IOException from
OpenExclusive(), hiding real filesystem errors; change the catch in
TryOpenExclusive to inspect the caught IOException (e.g., catch (IOException
ex)) and only return null for a true lock/sharing-violation (detect via
ex.HResult == 0x80070020 or a platform-appropriate check), otherwise rethrow the
exception so real permission/path errors propagate; update or add a small helper
(e.g., IsSharingViolation(IOException)) and use it in TryOpenExclusive to decide
between returning null and throwing.
In `@src/net-questdb-client/Qwp/Sf/QwpMmapSegment.cs`:
- Around line 376-387: The OffsetToFsn method can return a FSN before BaseFsn
when offset is less than the first envelope; update OffsetToFsn (which uses
_offsetTable.BinarySearch and BaseFsn) to defensively handle that case by
checking the computed idx after the ~idx - 1 adjustment and either clamp idx to
0 or throw a clear ArgumentOutOfRangeException (choose consistent behavior with
the rest of the class), and add a short comment documenting the precondition if
you opt to throw; ensure callers receive a valid FSN (>= BaseFsn) or a
descriptive exception.
In `@src/net-questdb-client/Qwp/Sf/QwpOrphanScanner.cs`:
- Around line 77-93: After acquiring the slot lock using
QwpSlotLock.TryAcquire(slotDir), re-check for the .failed sentinel
(FailedSentinel) before proceeding to drain: if
File.Exists(Path.Combine(slotDir, FailedSentinel)) then release/dispose slotLock
and continue; place this check immediately after the TryAcquire success and
before calling HasSegments(slotDir) so you close the TOCTOU window. Ensure you
call the slot lock's release/dispose method (slotLock.Dispose() or
slotLock.Release() per its API) when bailing.
In `@src/net-questdb-client/Qwp/Sf/QwpReconnectPolicy.cs`:
- Around line 87-97: The jitter function UniformDoubleJitter currently
multiplies the provided baseBackoff and can push a clamped backoff above
MaxBackoff; to fix, ensure jitter is applied to the clamped value (or clamp the
jittered result) so the final sleep never exceeds MaxBackoff: update
ComputeBackoff to call UniformDoubleJitter with the already-clamped backoff or
modify UniformDoubleJitter to accept a max and return Math.Min(jittered,
MaxBackoff); change references at UniformDoubleJitter and the ComputeBackoff
usage (also where schedule saturates around the lines noted) so the cap is
enforced after jittering.
In `@src/net-questdb-client/Senders/IQwpWebSocketSender.cs`:
- Around line 33-35: Update the stale XML remarks on the IQwpWebSocketSender
interface to reflect the current behavior: remove or reword the sentence stating
that ping/durable-ack support is "not yet shipped" and that methods may return
-1 or be no-ops, and instead document the actual current semantics (e.g., that
ping and durable-ack are supported and what their return values/exceptions are,
or note any specific supported/unsupported behaviors). Edit the <remarks> block
in IQwpWebSocketSender to mention the real behavior for
Ping/PersistentAck-related methods and reference the corresponding method names
(e.g., Ping, SendDurableAck) so callers aren’t misled.
In `@src/net-questdb-client/Senders/QwpWebSocketSender.cs`:
- Around line 734-755: The code treats the first non-durable OK as success even
when response.Sequence < sequence; change the logic in the loop in
QwpWebSocketSender (where response is handled) so that after confirming
response.IsOk you check whether response.Sequence >= sequence before clearing
awaitingAck or calling OnFlushSucceeded. If response.Sequence < sequence, treat
it as a stale ACK: call _inFlightWindow.AcknowledgeUpTo(response.Sequence) and
ProcessTableEntries(response.TableEntries, isDurable: false) and continue
waiting (do not set awaitingAck = false or call OnFlushSucceeded). Keep the
existing FailTerminal(response.ToException()) path unchanged for non-OK
responses.
In `@src/net-questdb-client/Utils/SenderOptions.cs`:
- Around line 172-176: EnsureValid() must enforce the same v1 sf_durability
invariant currently enforced in the constructor: inside the SenderOptions class
add a validation in EnsureValid() that checks the sf_durability/_sfDurability
value (and the protocol when relevant, e.g. ProtocolType.ws) and throws the same
IngressError(ErrorCode.ConfigError, $"`sf_durability` only accepts 'memory' in
v1, got `{_sfDurability}`") when a non-"memory" value is present; apply the same
change to the other validation block referenced (the section around lines
297-314) so object-initializers and with-expressions cannot bypass the transport
invariant.
- Around line 369-376: The WebSocket-only key list in SenderOptions.cs is
missing the new QWP auth keys, so add "token_x" and "token_y" to the
WebSocketOnlyKeys array (the same array used by ToString() and validation) so
non-WS transports will reject these options and ToString() will filter them out
for non-WebSocket SenderOptions instances; update the WebSocketOnlyKeys
declaration to include the exact strings "token_x" and "token_y".
---
Nitpick comments:
In `@src/example-websocket-auth-tls/Program.cs`:
- Around line 20-23: The example currently hardcodes credentials and disables
TLS verification in the Sender.New connection string; replace that literal with
environment-driven values and a safe default for TLS verification: read
username, password, host/port and tls_verify from environment variables (e.g.,
via Environment.GetEnvironmentVariable) and fall back to secure defaults
(tls_verify=on) if not provided, and ensure Sender.New is passed the constructed
connection string rather than the hardcoded one so unsafe settings are explicit
at runtime; update any comments to instruct users to set ENV vars for testing
instead of copying credentials into source.
In `@src/net-questdb-client-benchmarks/BenchSfAppend.cs`:
- Around line 86-100: The Cleanup method currently swallows all exceptions
(empty catch blocks) for _sfSender.Dispose(), _wsSender.Dispose(),
_server.DisposeAsync(), and Directory.Delete(_sfRoot), which can hide cleanup
failures; replace each empty catch with a catch (Exception ex) that logs the
exception at debug level (e.g., _logger.Debug/LogDebug or
Console.Error.WriteLine if no logger exists) including context (which resource
failed) so failures in Dispose/DisposeAsync and Directory.Delete are recorded;
ensure you capture exceptions for the asynchronous server dispose (await inside
try/catch around _server.DisposeAsync()) and include the resource name
(_sfSender, _wsSender, _server, _sfRoot) in each log message.
In `@src/net-questdb-client-tests/Qwp/QwpEncoderTests.cs`:
- Around line 582-592: The method FindFirstColumnDataOffset currently takes
unused parameters frame and userColCount which are immediately discarded; either
remove these parameters from the signature and all call sites (leaving
tableNameLen and userColDefSize only), or keep them and replace the discard
lines with an explicit clarifying comment like "// TODO: reserved for future
multi-column encoder validation" and remove the "_ = frame;" since it's unused;
update any references to userColCount/frame across tests to match the new
signature if you choose removal, or leave them in place but document intent if
you choose to retain them.
In `@src/net-questdb-client-tests/Qwp/QwpVarintTests.cs`:
- Around line 140-158: The test RoundTrip_RandomFuzz_PreservesValue currently
builds a 64-bit value with (hi << 32) | lo which discards the upper 32 bits of
hi; instead generate a full random ulong before writing/reading. Update the
value generation in the test (inside RoundTrip_RandomFuzz_PreservesValue) to
produce a true 64-bit unsigned value—e.g. fill an 8-byte stackalloc buffer with
rnd.NextBytes and use BinaryPrimitives.ReadUInt64LittleEndian, or combine two
32-bit Next() calls correctly—to ensure QwpVarint.Write/ QwpVarint.Read are
exercised across the full 64-bit range.
- Around line 79-89: The test RoundTrip_PowerOfTwoBoundaries_PreservesValue uses
a redundant ternary to set value (bit == 63 ? 1ul << 63 : 1ul << bit); simplify
by replacing that expression with the single safe shift 1ul << bit and keep the
surrounding logic (the loop over bit and calls to AssertRoundTrip and the
conditional AssertRoundTrip(value - 1)) unchanged so behavior is identical but
cleaner.
In `@src/net-questdb-client-tests/Qwp/QwpWebSocketSenderTests.cs`:
- Around line 556-562: Extract the repeated SF cleanup block into a single
helper method (e.g., add a private static TryDeleteDirectory(string path) in the
QwpWebSocketSenderTests class) that performs the current pattern (check
Directory.Exists, then try Directory.Delete(path, recursive: true) with an empty
catch), and replace each inline occurrence like the one using sfRoot with a call
to TryDeleteDirectory(sfRoot); this reduces duplication and keeps behavior
identical.
In `@src/net-questdb-client/Qwp/QwpColumn.cs`:
- Around line 737-749: EnsureOffsetCapacity currently dereferences StrOffsets
(StrOffsets!) and can NRE if called before allocation; modify
EnsureOffsetCapacity to defensively handle null by initializing StrOffsets to a
sensible starting array (e.g., length 1 or requiredCount rounded up to
power-of-two) before running the resize loop, and then proceed with the existing
doubling logic to grow to requiredCount; reference EnsureOffsetCapacity and the
StrOffsets field (and note AppendVarchar currently initializes it) so the change
is focused and keeps existing behavior for current callers.
In `@src/net-questdb-client/Qwp/QwpInFlightWindow.cs`:
- Around line 196-237: The AwaitEmpty method currently uses a poll loop with
CancellationPollMs and Monitor.Wait(_lock, waitMs) to observe cancellation,
causing extra wakeups and latency; instead, register ct.Register(() => {
lock(_lock) Monitor.PulseAll(_lock); }) when ct.CanBeCanceled and, when no
deadline is set (hasDeadline == false), call Monitor.Wait(_lock) (indefinite
wait) so the cancellation callback will wake the waiter; ensure the registration
is disposed/unregistered before returning or throwing and preserve existing
timeout behavior when hasDeadline is true.
In `@src/net-questdb-client/Qwp/Sf/QwpBackgroundDrainerPool.cs`:
- Around line 146-156: The code calls Task.WhenAll(snapshot) twice; instead
capture the Task returned from the first Task.WhenAll into a local (e.g.
whenAllTask) and reuse it after invoking SfCleanup.Run(() =>
_shutdownCts.Cancel()) instead of calling Task.WhenAll(snapshot) again; update
the variable currently assigned to allJoined to Wait(TimeSpan.FromSeconds(2)) on
that stored whenAllTask (and catch exceptions as before) so you don't
allocate/duplicate the same aggregate Task for the same snapshot array.
- Around line 100-106: The continuation created in QwpBackgroundDrainerPool (the
task.ContinueWith block that locks _trackingLock and calls
_runningTasks.Remove(t)) should be executed synchronously to avoid an
unnecessary thread-pool hop; modify the ContinueWith call to include
TaskContinuationOptions.ExecuteSynchronously (keep the current
TaskScheduler.Default or omit scheduler if not needed) so the lightweight
cleanup runs inline when possible.
In `@src/net-questdb-client/Qwp/Sf/QwpMmapSegment.cs`:
- Around line 282-310: The Dispose method in QwpMmapSegment.cs doesn't call
GC.SuppressFinalize(this); add a call to GC.SuppressFinalize(this) (e.g. at the
end of Dispose after setting _disposed = true) to avoid unnecessary finalization
overhead when consumers dispose the object; no changes to _view/_mmap cleanup
logic are needed unless you add a finalizer, in which case ensure the finalizer
and Dispose(bool) pattern are implemented correctly.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: 1bd55a87-b27a-4fd1-a783-8fbfe3fe17b1
📒 Files selected for processing (75)
README.mddocs/qwip-benchmarks.mdexamples.manifest.yamlnet-questdb-client.slnsrc/dummy-http-server/DummyQwpServer.cssrc/example-websocket-auth-tls/Program.cssrc/example-websocket-auth-tls/example-websocket-auth-tls.csprojsrc/example-websocket/Program.cssrc/example-websocket/example-websocket.csprojsrc/net-questdb-client-benchmarks/BenchInsertsWs.cssrc/net-questdb-client-benchmarks/BenchLatencyWs.cssrc/net-questdb-client-benchmarks/BenchSfAppend.cssrc/net-questdb-client-benchmarks/BenchSfThroughput.cssrc/net-questdb-client-benchmarks/Program.cssrc/net-questdb-client-tests/QuestDbManager.cssrc/net-questdb-client-tests/QuestDbWebSocketIntegrationTests.cssrc/net-questdb-client-tests/Qwp/QwpColumnExtendedTypesTests.cssrc/net-questdb-client-tests/Qwp/QwpColumnTests.cssrc/net-questdb-client-tests/Qwp/QwpEncoderTests.cssrc/net-questdb-client-tests/Qwp/QwpGorillaTests.cssrc/net-questdb-client-tests/Qwp/QwpInFlightWindowTests.cssrc/net-questdb-client-tests/Qwp/QwpResponseTests.cssrc/net-questdb-client-tests/Qwp/QwpSchemaCacheTests.cssrc/net-questdb-client-tests/Qwp/QwpSymbolDictionaryTests.cssrc/net-questdb-client-tests/Qwp/QwpTableBufferTests.cssrc/net-questdb-client-tests/Qwp/QwpVarintTests.cssrc/net-questdb-client-tests/Qwp/QwpWebSocketSenderTests.cssrc/net-questdb-client-tests/Qwp/QwpWebSocketTransportTests.cssrc/net-questdb-client-tests/Qwp/Sf/QwpBackgroundDrainerPoolTests.cssrc/net-questdb-client-tests/Qwp/Sf/QwpBackgroundDrainerTests.cssrc/net-questdb-client-tests/Qwp/Sf/QwpCrc32CTests.cssrc/net-questdb-client-tests/Qwp/Sf/QwpCursorSendEngineTests.cssrc/net-questdb-client-tests/Qwp/Sf/QwpFilesTests.cssrc/net-questdb-client-tests/Qwp/Sf/QwpMmapSegmentTests.cssrc/net-questdb-client-tests/Qwp/Sf/QwpOrphanScannerTests.cssrc/net-questdb-client-tests/Qwp/Sf/QwpReconnectPolicyTests.cssrc/net-questdb-client-tests/Qwp/Sf/QwpSegmentManagerTests.cssrc/net-questdb-client-tests/Qwp/Sf/QwpSegmentRingTests.cssrc/net-questdb-client-tests/Qwp/Sf/QwpSlotLockTests.cssrc/net-questdb-client-tests/SenderOptionsTests.cssrc/net-questdb-client/Enums/ProtocolType.cssrc/net-questdb-client/Enums/QwpStatusCode.cssrc/net-questdb-client/Enums/QwpTypeCode.cssrc/net-questdb-client/Qwp/QwpBitWriter.cssrc/net-questdb-client/Qwp/QwpColumn.cssrc/net-questdb-client/Qwp/QwpConstants.cssrc/net-questdb-client/Qwp/QwpEncoder.cssrc/net-questdb-client/Qwp/QwpException.cssrc/net-questdb-client/Qwp/QwpGorilla.cssrc/net-questdb-client/Qwp/QwpInFlightWindow.cssrc/net-questdb-client/Qwp/QwpResponse.cssrc/net-questdb-client/Qwp/QwpSchemaCache.cssrc/net-questdb-client/Qwp/QwpSymbolDictionary.cssrc/net-questdb-client/Qwp/QwpTableBuffer.cssrc/net-questdb-client/Qwp/QwpVarint.cssrc/net-questdb-client/Qwp/QwpWebSocketTransport.cssrc/net-questdb-client/Qwp/Sf/IQwpCursorTransport.cssrc/net-questdb-client/Qwp/Sf/IQwpSlotDrainer.cssrc/net-questdb-client/Qwp/Sf/QwpBackgroundDrainer.cssrc/net-questdb-client/Qwp/Sf/QwpBackgroundDrainerPool.cssrc/net-questdb-client/Qwp/Sf/QwpCrc32C.cssrc/net-questdb-client/Qwp/Sf/QwpCursorSendEngine.cssrc/net-questdb-client/Qwp/Sf/QwpFiles.cssrc/net-questdb-client/Qwp/Sf/QwpMmapSegment.cssrc/net-questdb-client/Qwp/Sf/QwpOrphanScanner.cssrc/net-questdb-client/Qwp/Sf/QwpReconnectPolicy.cssrc/net-questdb-client/Qwp/Sf/QwpSegmentManager.cssrc/net-questdb-client/Qwp/Sf/QwpSegmentRing.cssrc/net-questdb-client/Qwp/Sf/QwpSlotLock.cssrc/net-questdb-client/Qwp/Sf/SfCleanup.cssrc/net-questdb-client/Sender.cssrc/net-questdb-client/Senders/IQwpWebSocketSender.cssrc/net-questdb-client/Senders/QwpWebSocketSender.cssrc/net-questdb-client/Utils/SenderOptions.cssrc/net-questdb-client/net-questdb-client.csproj
There was a problem hiding this comment.
Actionable comments posted: 15
♻️ Duplicate comments (1)
src/net-questdb-client/Utils/SenderOptions.cs (1)
339-364:⚠️ Potential issue | 🟠 MajorGate
token_x/token_yon the WS path too.Neither
ValidateWebSocketKeysAgainstDefaults()norWebSocketOnlyKeysincludes the new auth pair, so non-WS callers can still set them via config strings, object initializers, orwithexpressions and passEnsureValid().🔧 Minimal fix
if (_requestDurableAck != defaults._requestDurableAck) Throw(nameof(request_durable_ack)); + if (_tokenX != defaults._tokenX) Throw(nameof(token_x)); + if (_tokenY != defaults._tokenY) Throw(nameof(token_y)); if (_sfDir != defaults._sfDir) Throw(nameof(sf_dir));private static readonly string[] WebSocketOnlyKeys = { "in_flight_window", "close_timeout", "max_schemas_per_connection", "gorilla", "request_durable_ack", "sf_dir", "sender_id", "sf_max_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", - "drain_orphans", "max_background_drainers", + "drain_orphans", "max_background_drainers", "token_x", "token_y", };As per coding guidelines
src/net-questdb-client/Utils/SenderOptions.cs: Config string parsing and sender creation must route through SenderOptions.EnsureValid() which validates auth combinations, TLS settings, multi-addr restrictions, gzip restrictions on WS, and WS-only-keys heuristics before protocol dispatch.Also applies to: 392-399
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/net-questdb-client/Utils/SenderOptions.cs` around lines 339 - 364, ValidateWebSocketKeysAgainstDefaults and the WebSocketOnlyKeys set are missing the new auth pair (token_x / token_y), allowing non-WS callers to set them; update ValidateWebSocketKeysAgainstDefaults to compare _tokenX and _tokenY (or the exact field names used) against a fresh SenderOptions() defaults and call Throw(nameof(token_x)) / Throw(nameof(token_y)) when they differ, and add the corresponding token_x and token_y identifiers to the WebSocketOnlyKeys collection; also mirror the same additions in the other validation block referenced (around the second region ~392-399) so EnsureValid() enforces these keys are WS-only before protocol dispatch.
🧹 Nitpick comments (1)
README.md (1)
271-271: Use “scheme” instead of “schema” forprotocol.
protocolrefers to URI scheme (http,tcp,ws), not schema. Small wording fix for clarity.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@README.md` at line 271, Update the README text labeling for the `protocol` field: change the parenthetical "schema" to "scheme" so the entry reads `protocol` (scheme) and optionally clarify that it refers to the URI scheme (e.g., http, tcp, ws) in the same table row where `protocol` is documented.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@README.md`:
- Around line 306-307: Update the README table so the documented default values
match the runtime defaults defined in SenderOptions.cs (lines referencing
sf_max_bytes, sf_max_total_bytes and reconnect_max_backoff_millis): change
sf_max_bytes from 67108864 to 4194304 (4 MiB); make sf_max_total_bytes reflect
the conditional default (10 GiB when sf_dir is set, otherwise 128 MiB) rather
than long.MaxValue; and change reconnect_max_backoff_millis from 30000 to 1000;
ensure the option names in the table (sf_max_bytes, sf_max_total_bytes,
reconnect_max_backoff_millis) exactly match those used in SenderOptions.cs so
readers see the runtime-consistent defaults.
- Line 166: Resolve the contradictory guidance by making both mentions of
in_flight_window consistent: either state that in_flight_window must be >=2
everywhere (and change the sync-semantics recommendation on the second
occurrence to recommend in_flight_window=2 for sync-like behavior), or
explicitly note that in_flight_window=1 is only allowed on non-WebSocket
transports and that the WebSocket transport rejects 1; update the two
occurrences referencing in_flight_window and the WebSocket/sync semantics
wording to match that single intended rule (symbols to edit: in_flight_window,
WebSocket, "sync semantics" mention).
In `@src/net-questdb-client-tests/Qwp/QwpInFlightWindowTests.cs`:
- Around line 117-124: The test AwaitEmpty_NotEmpty_TimesOut uses a very short
50ms timeout which flakes under CI; update this test to use a more generous
timeout (e.g., TimeSpan.FromMilliseconds(200)) when calling
QwpInFlightWindow.AwaitEmpty, and refactor the related cancellation/timeout
tests below to reuse a shared timeout variable (e.g., var timeout =
TimeSpan.FromMilliseconds(200)) so all AwaitEmpty(TimeSpan) assertions use the
same larger margin.
- Around line 161-179: The test can race because AcknowledgeUpTo(1) may run
before AwaitEmpty(...) is actually blocked; fix by starting the waiter on a Task
(call AwaitEmpty in a Task.Run), then assert the Task is still pending (e.g.,
Task.IsCompleted == false or Wait with a short timeout returns false) to prove
the wait path is exercised, only then signal the producerEnteredAwait and call
AcknowledgeUpTo(1); finally wait for the waiter Task to complete and assert
w.IsEmpty. Ensure you reference AwaitEmpty, AcknowledgeUpTo,
producerEnteredAwait/ManualResetEventSlim, and the waiter Task in the updated
test.
In `@src/net-questdb-client/Qwp/QwpEncoder.cs`:
- Around line 123-138: The current loop in QwpEncoder (calling WriteTableBlock
which uses schemaCache.PrepareSchema/table.SchemaId) can advance the schema
cache before we know the frame will be sendable; if payloadLength exceeds
MaxBatchBytes we must not leave schema ids advanced. Change the flow so
schema-cache mutations are staged and only committed after the size check:
either (a) add a dry-run PrepareSchema variant or a PrepareSchemaCollect method
that returns the schema-id and size delta without mutating schemaCache,
accumulate sizes during the loop, then if payloadLength <= MaxBatchBytes call a
CommitPreparedSchemas method to apply the collected PrepareSchema results (or
call the real PrepareSchema), or (b) record all schema-cache and table.SchemaId
changes made by WriteTableBlock into a temp list and, on oversize, iterate that
list to roll them back before throwing. Touch points: WriteTableBlock,
schemaCache.PrepareSchema, table.SchemaId, and the payload-size check — ensure
only CommitPreparedSchemas (or no-op rollback) advances
maxSentSchemaId/maxSentSymbolId for the appropriate modes (Sync/Async/SF) as per
existing mode semantics.
In `@src/net-questdb-client/Qwp/QwpSchemaCache.cs`:
- Around line 90-101: PrepareSchema currently increments and assigns
_nextSchemaId and updates _maxSentSchemaId at encode time (setting
table.SchemaId and returning SchemaModeFull), which must be moved out of this
mode-agnostic path; instead leave PrepareSchema to only decide mode/need and
assign a provisional id without advancing global counters, and implement
mode-specific advancement: in the ACK handling path for Sync mode (where server
ACKs are processed) advance _maxSentSchemaId and commit the provisional
table.SchemaId only after ACK; in the Async enqueue path advance
_nextSchemaId/_maxSentSchemaId immediately upon successful enqueue (with the
same terminal-error guard currently used), and for SF/self-sufficient frames do
not advance _maxSentSchemaId at all; keep references to _nextSchemaId,
_maxSentSchemaId, _maxSchemasPerConnection, PrepareSchema, table.SchemaId and
QwpConstants.SchemaModeFull to locate and move the increment logic.
In `@src/net-questdb-client/Qwp/QwpWebSocketTransport.cs`:
- Around line 270-289: ReadNegotiatedVersion currently returns
QwpConstants.SupportedIngestVersion when the X-QWP-Version header exists but its
value cannot be parsed; change this to fail fast by throwing
ProtocolVersionError instead of silently falling back. In ReadNegotiatedVersion
(referencing _client.HttpResponseHeaders and QwpConstants.HeaderVersion) detect
the presence of the header(s) and, if any value is present but int.TryParse
fails for all values, throw ProtocolVersionError with a clear message about the
unparsable X-QWP-Version; only fall back to QwpConstants.SupportedIngestVersion
when the header is entirely absent (headers is null or TryGetValue returns
false).
In `@src/net-questdb-client/Qwp/Sf/QwpCursorSendEngine.cs`:
- Around line 721-733: FireAppendSignalLocked and FireAckSignalLocked currently
call prev.TrySetResult(true) synchronously while the caller holds _stateLock;
change each to bounce the completion off the lock-holder's stack by invoking
Task.Run(() => prev.TrySetResult(true)) instead of calling TrySetResult inline.
Update FireAppendSignalLocked and FireAckSignalLocked to capture prev as before,
replace the direct TrySetResult call with Task.Run(() =>
prev.TrySetResult(true)), and keep the existing NewSignal assignment; this
ensures callers like SetTerminal, AppendBlockingSlow, and the spare-installed
callback do not execute continuation work under the lock.
In `@src/net-questdb-client/Qwp/Sf/QwpMmapSegment.cs`:
- Around line 205-207: TryAppend mutates the shared List<int> _offsetTable while
OffsetOfEnvelope/OffsetToFsn and QwpSegmentRing.TryReadFrame read it
unsynchronized, causing race/crash; fix by introducing a dedicated lock (e.g.,
private readonly object _offsetTableLock) on the segment and wrap all
accesses/modifications of _offsetTable inside lock(_offsetTableLock) in
TryAppend, OffsetOfEnvelope, OffsetToFsn and any external call sites (or make
OffsetOfEnvelope/OffsetToFsn take the lock internally), and update
QwpSegmentRing.TryReadFrame so it either holds the segment's _offsetTableLock
while calling seg.OffsetOfEnvelope()/seg.TryReadFrame() or relies on those
methods to lock internally to ensure safe concurrent reads during producer
appends.
In `@src/net-questdb-client/Qwp/Sf/QwpSegmentRing.cs`:
- Around line 499-503: EnsureNotClosed currently acquires _lock on every
producer hot-path call (from TryAppend), serializing appends; change it to a
lock-free disposal check by making _closed a volatile/atomic field (e.g.,
volatile int or bool) and replace the lock in EnsureNotClosed with a
Volatile.Read/Interlocked-based read of _closed, throwing
ObjectDisposedException(nameof(QwpSegmentRing)) when true; keep any
manager-thread synchronization for rotation/trim separate so the manager
pre-provisions hot spares and producers never block on segment allocation.
In `@src/net-questdb-client/Senders/QwpWebSocketSender.cs`:
- Around line 939-962: The ACK-processing block in QwpWebSocketSender (inside
the receive loop handling QwpResponse.Parse results) can throw (e.g.,
_inFlightWindow.AcknowledgeUpTo, _slot.Release, ProcessTableEntries) and
currently bubbles as a fault without setting _terminalError; wrap the entire
section that executes after QwpResponse.Parse (including handling of
response.IsDurableAck, response.IsOk, _inFlightWindow.AcknowledgeUpTo,
_slot.Release, and ProcessTableEntries) in a try/catch that calls
FailTerminal(ex) and returns so protocol/ACK errors are terminalized immediately
and _terminalError is set.
- Around line 784-835: The call to OnFlushSucceeded() is happening too early (in
the method that calls EncodeFrameInto(idx)), causing the batch and symbol
dictionary to be cleared before the frame is actually reserved/enqueued; move
the OnFlushSucceeded() call to after the enqueue is guaranteed to succeed (i.e.,
after awaiting _slot.WaitAsync(linkedCt), adding the sequence with
_inFlightWindow.Add(seq), and after _sendChannel!.Writer.TryWrite(new
AsyncBatch(idx, frame)) returns true). Ensure error paths that catch
OperationCanceledException or generic Exception still release _slot and
_encoderReady[idx] and do not call OnFlushSucceeded(); only invoke
OnFlushSucceeded() once TryWrite has succeeded (and handle the failure branch by
calling FailTerminal/throwing as currently done).
- Around line 969-977: The methods QwpWebSocketSender.CancelRow() and
QwpWebSocketSender.Truncate() currently silently no-op which violates the
ISender contract; instead update both methods in the QwpWebSocketSender class to
immediately throw a clear exception (choose NotSupportedException or
InvalidApiCall) explaining that cancel/ truncate are not supported for the
WebSocket multi-table columnar sender because QwpTableBuffer does not expose
those operations; ensure the exception message references the method name
(CancelRow/Truncate) and that callers receive the thrown exception rather than a
silent no-op.
In `@src/net-questdb-client/Utils/SenderOptions.cs`:
- Around line 179-182: The default for sf_max_total_bytes is being chosen based
on _sfDir only in the connection-string parsing path, causing instances created
via object initializer to keep the 128MiB initializer; fix by computing the
defaultMaxTotal at the time options are finalized (or when parsing is executed)
instead of relying on the field initializer: move the logic that sets
defaultMaxTotal (the ternary that checks _sfDir and picks 128MiB vs 10GiB) into
the same initialization/finalization code path that calls ParseLongWithDefault
(i.e., the method that currently calls
ParseLongWithDefault(nameof(sf_max_total_bytes), ...)), or call that computation
after sf_dir is set (e.g., in the SenderOptions constructor or in the sf_dir
property setter) so ParseLongWithDefault writes the correct _sfMaxTotalBytes
regardless of how SenderOptions was created.
- Around line 156-166: The constructor currently calls ParseStringWithDefault
for token but never for token_x/token_y so _tokenX and _tokenY remain unset; add
ParseStringWithDefault(nameof(token_x), null, out _tokenX) and
ParseStringWithDefault(nameof(token_y), null, out _tokenY) into SenderOptions
initialization (same block where ParseStringWithDefault(nameof(token), ...) is
called) so connection-string keys token_x/token_y are parsed; ensure these
fields are then validated by SenderOptions.EnsureValid() alongside existing auth
logic.
---
Duplicate comments:
In `@src/net-questdb-client/Utils/SenderOptions.cs`:
- Around line 339-364: ValidateWebSocketKeysAgainstDefaults and the
WebSocketOnlyKeys set are missing the new auth pair (token_x / token_y),
allowing non-WS callers to set them; update ValidateWebSocketKeysAgainstDefaults
to compare _tokenX and _tokenY (or the exact field names used) against a fresh
SenderOptions() defaults and call Throw(nameof(token_x)) /
Throw(nameof(token_y)) when they differ, and add the corresponding token_x and
token_y identifiers to the WebSocketOnlyKeys collection; also mirror the same
additions in the other validation block referenced (around the second region
~392-399) so EnsureValid() enforces these keys are WS-only before protocol
dispatch.
---
Nitpick comments:
In `@README.md`:
- Line 271: Update the README text labeling for the `protocol` field: change the
parenthetical "schema" to "scheme" so the entry reads `protocol` (scheme) and
optionally clarify that it refers to the URI scheme (e.g., http, tcp, ws) in the
same table row where `protocol` is documented.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: aa301fc4-8e82-455d-9c54-9499530c0028
📒 Files selected for processing (39)
CLAUDE.mdREADME.mddocs/qwp-benchmarks.mdsrc/net-questdb-client-tests/QuestDbWebSocketIntegrationTests.cssrc/net-questdb-client-tests/Qwp/QwpColumnExtendedTypesTests.cssrc/net-questdb-client-tests/Qwp/QwpEncoderTests.cssrc/net-questdb-client-tests/Qwp/QwpGorillaTests.cssrc/net-questdb-client-tests/Qwp/QwpInFlightWindowTests.cssrc/net-questdb-client-tests/Qwp/QwpResponseTests.cssrc/net-questdb-client-tests/Qwp/QwpSchemaCacheTests.cssrc/net-questdb-client-tests/Qwp/QwpTableBufferTests.cssrc/net-questdb-client-tests/Qwp/QwpVarintTests.cssrc/net-questdb-client-tests/Qwp/QwpWebSocketSenderTests.cssrc/net-questdb-client-tests/Qwp/Sf/QwpCursorSendEngineTests.cssrc/net-questdb-client-tests/Qwp/Sf/QwpMmapSegmentTests.cssrc/net-questdb-client-tests/Qwp/Sf/QwpSegmentManagerTests.cssrc/net-questdb-client-tests/Qwp/Sf/QwpSegmentRingTests.cssrc/net-questdb-client-tests/SenderOptionsTests.cssrc/net-questdb-client/Qwp/QwpBitWriter.cssrc/net-questdb-client/Qwp/QwpColumn.cssrc/net-questdb-client/Qwp/QwpConstants.cssrc/net-questdb-client/Qwp/QwpEncoder.cssrc/net-questdb-client/Qwp/QwpInFlightWindow.cssrc/net-questdb-client/Qwp/QwpResponse.cssrc/net-questdb-client/Qwp/QwpSchemaCache.cssrc/net-questdb-client/Qwp/QwpSymbolDictionary.cssrc/net-questdb-client/Qwp/QwpTableBuffer.cssrc/net-questdb-client/Qwp/QwpWebSocketTransport.cssrc/net-questdb-client/Qwp/Sf/QwpBackgroundDrainerPool.cssrc/net-questdb-client/Qwp/Sf/QwpCursorSendEngine.cssrc/net-questdb-client/Qwp/Sf/QwpMmapSegment.cssrc/net-questdb-client/Qwp/Sf/QwpOrphanScanner.cssrc/net-questdb-client/Qwp/Sf/QwpReconnectPolicy.cssrc/net-questdb-client/Qwp/Sf/QwpSegmentManager.cssrc/net-questdb-client/Qwp/Sf/QwpSegmentRing.cssrc/net-questdb-client/Senders/IQwpWebSocketSender.cssrc/net-questdb-client/Senders/ISender.cssrc/net-questdb-client/Senders/QwpWebSocketSender.cssrc/net-questdb-client/Utils/SenderOptions.cs
✅ Files skipped from review due to trivial changes (4)
- CLAUDE.md
- docs/qwp-benchmarks.md
- src/net-questdb-client-tests/Qwp/Sf/QwpSegmentRingTests.cs
- src/net-questdb-client/Qwp/QwpConstants.cs
🚧 Files skipped from review as they are similar to previous changes (8)
- src/net-questdb-client/Senders/IQwpWebSocketSender.cs
- src/net-questdb-client/Qwp/Sf/QwpOrphanScanner.cs
- src/net-questdb-client/Qwp/Sf/QwpReconnectPolicy.cs
- src/net-questdb-client-tests/QuestDbWebSocketIntegrationTests.cs
- src/net-questdb-client-tests/Qwp/Sf/QwpMmapSegmentTests.cs
- src/net-questdb-client-tests/Qwp/Sf/QwpSegmentManagerTests.cs
- src/net-questdb-client/Qwp/QwpBitWriter.cs
- src/net-questdb-client/Qwp/QwpSymbolDictionary.cs
There was a problem hiding this comment.
Actionable comments posted: 5
🧹 Nitpick comments (4)
README.md (1)
150-167: Consider tightening the “async-only / in_flight_window=1 rejected” wording for clarity.Line 166 states the WebSocket transport is async-only and that
in_flight_window=1is rejected. That aligns with the table at Line 299, but “async-only” can be interpreted as “Sync APIs don’t work at all” vs “Sync mode is mapped/blocked by IFW validation”.No change required, but please ensure the language elsewhere (especially around
SendvsSendAsyncand disposal behavior) matches the actual enforced rule, not just the config validation rule.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@README.md` around lines 150 - 167, The README currently says "WebSocket transport is async-only — in_flight_window=1 is rejected", which is ambiguous; change the wording to state that the WebSocket transport requires using the async APIs (SendAsync, AtAsync, and async disposal) whenever in_flight_window > 1 because the sender pipelines frames asynchronously, and that an explicit config of in_flight_window=1 is rejected by validation; clarify that the synchronous Send path remains present but is effectively blocked/mapped by IFW validation for WebSocket transport, so callers must use Sender.New(...)/AtAsync(...) and await SendAsync() or async disposal to ensure outstanding frames are flushed.src/net-questdb-client-tests/Qwp/QwpBitWriterTests.cs (1)
83-93: PreferAssert.Throwsover manual try/catch for exception assertions.This is more idiomatic NUnit and improves failure output readability. The codebase already uses this pattern throughout the test suite.
♻️ Optional refactor
- InvalidOperationException? thrown = null; - try { w.WriteBits(0, 1); } - catch (InvalidOperationException ex) { thrown = ex; } - Assert.That(thrown, Is.Not.Null); + Assert.Throws<InvalidOperationException>(() => w.WriteBits(0, 1));🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/net-questdb-client-tests/Qwp/QwpBitWriterTests.cs` around lines 83 - 93, Replace the manual try/catch in the test method WriteBits_ExhaustedBuffer_Throws with NUnit's Assert.Throws pattern: call Assert.Throws<InvalidOperationException>(() => w.WriteBits(0, 1)) to assert the exception from QwpBitWriter.WriteBits and remove the thrown variable/try-catch block so the test uses the standard idiomatic assertion used elsewhere in the suite.src/net-questdb-client/Qwp/QwpInFlightWindow.cs (1)
29-48: Trim comment blocks to non-obvious “why” only.The class-level XML docs are largely behavioral restatements; please keep comments focused on hidden constraints/counter-intuitive decisions and drop descriptive narration.
As per coding guidelines: "
**/*.cs: Add comments only when explaining the 'why' behind non-obvious logic (hidden constraints, counter-intuitive ordering, runtime workarounds); do not restate what code does or reference plan documents".🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/net-questdb-client/Qwp/QwpInFlightWindow.cs` around lines 29 - 48, The class XML doc for QwpInFlightWindow is too verbose and restates behavior; trim it to a short "why" note that highlights only non-obvious constraints/decisions (e.g., sentinel choice of -1 for AckedSequence and HighestSentSequence to disambiguate "never" vs sequence 0, the cumulative-ACK semantics that AcknowledgeUpTo absorbs lower/out-of-order seqs and throws on seq > HighestSentSequence as a server bug, and the terminal-failure semantics where FailAll records the first failure rethrown by AwaitEmpty). Remove procedural/descriptive narration about how the class works and keep only these hidden constraints and counter-intuitive choices referenced by their symbols (QwpInFlightWindow, AckedSequence, HighestSentSequence, AcknowledgeUpTo, FailAll, AwaitEmpty).src/net-questdb-client-tests/Qwp/QwpWebSocketSenderTests.cs (1)
259-265: Avoid depending on port 1 being closed.This is environment-dependent and can turn into a flaky CI failure. Prefer reserving an ephemeral local port, closing it, then connecting to that known-unused port.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/net-questdb-client-tests/Qwp/QwpWebSocketSenderTests.cs` around lines 259 - 265, The test ConnectFailure_ClosedPort_RaisesIngressError currently assumes port 1 is closed; instead reserve an ephemeral local port and use that to guarantee a closed/unused port: in the test (method ConnectFailure_ClosedPort_RaisesIngressError) create a TcpListener bound to IPAddress.Loopback with port 0 to obtain an available port, read the assigned port, stop/close the listener to free that port, then construct the Sender.New connection string using the obtained port (e.g. "ws::addr=127.0.0.1:{port};auto_flush=off;") and assert the IngressError with ErrorCode.SocketError as before.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@src/net-questdb-client-tests/Qwp/QwpInFlightWindowTests.cs`:
- Around line 169-174: The test currently calls
waitTask.Wait(TimeSpan.FromSeconds(2)) but ignores its boolean result; change
the assertion to verify the waiter actually completed after the ACK by asserting
the Wait call returned true (or assert waitTask.IsCompleted) immediately after
w.AcknowledgeUpTo(1) so the test fails if the waiter timed out; keep the
existing subsequent Assert.That(w.IsEmpty) and reference the waiter task
(waitTask), the AwaitEmpty call, and AcknowledgeUpTo(1) when locating where to
add the boolean check.
In `@src/net-questdb-client-tests/Qwp/QwpWebSocketSenderTests.cs`:
- Around line 267-274: The test InFlightWindow_One_Rejected currently expects
Sender.New("...in_flight_window=1...") to throw a ConfigError, but
in_flight_window=1 represents the synchronous WebSocket mode and should be
accepted; update the test to create the Sender via Sender.New without expecting
an exception and assert the sender enters sync mode (e.g., verify creation
succeeds and any public property or behavior indicating Sync mode is set)
instead of asserting ErrorCode.ConfigError or an error message mentioning
"in_flight_window". Ensure references to Sender.New and the in_flight_window=1
setting are used to locate and change the test.
In `@src/net-questdb-client/Qwp/QwpWebSocketTransport.cs`:
- Around line 129-140: The catch block should treat any non-101 HTTP upgrade
response as terminal instead of only 401/403: inspect _client.HttpStatusCode and
if it is not 101 throw an IngressError with ErrorCode.AuthError (including the
original exception ex and _options.Uri) so the SF cursor engine skips
reconnects; otherwise keep the existing fallback that throws
IngressError(ErrorCode.SocketError, ...) for non-HTTP-upgrade failures.
In `@src/net-questdb-client/Qwp/Sf/QwpCursorSendEngine.cs`:
- Around line 466-485: The connect path can throw ProtocolVersionError from
QwpWebSocketTransport.ConnectAsync() before RunConnectionAsync() starts but the
current catch only treats AuthError as terminal, causing endless retries; update
the exception handling in QwpCursorSendEngine.RunConnectionAsync (the
catch(Exception ex) block that follows the AuthError handler) to treat
ProtocolVersionError (and any other protocol/version-related exceptions thrown
by QwpWebSocketTransport.ConnectAsync) as terminal by calling SetTerminal(ex)
and returning, rather than retrying; locate the catch blocks around
SetTerminal/BackoffOrGiveUpAsync and add a branch checking for
ProtocolVersionError (or the transport's specific exception type) to call
SetTerminal and exit.
- Around line 336-375: FlushAsync in QwpCursorSendEngine computes deadline =
DateTime.MaxValue for Timeout.InfiniteTimeSpan which makes remaining exceed
Timer.MaxSupportedTimeout and causes Task.WaitAsync(remaining,
cancellationToken) to throw; update FlushAsync to detect the infinite-timeout
case and call the no-timeout overload await
waitTask.WaitAsync(cancellationToken).ConfigureAwait(false) (or otherwise avoid
passing an oversized TimeSpan) when timeout == Timeout.InfiniteTimeSpan, and
only compute and pass remaining to WaitAsync when timeout is finite; refer to
the FlushAsync method, _ackSignal.Task waitTask, and the timeout/remaining logic
to implement the branch.
---
Nitpick comments:
In `@README.md`:
- Around line 150-167: The README currently says "WebSocket transport is
async-only — in_flight_window=1 is rejected", which is ambiguous; change the
wording to state that the WebSocket transport requires using the async APIs
(SendAsync, AtAsync, and async disposal) whenever in_flight_window > 1 because
the sender pipelines frames asynchronously, and that an explicit config of
in_flight_window=1 is rejected by validation; clarify that the synchronous Send
path remains present but is effectively blocked/mapped by IFW validation for
WebSocket transport, so callers must use Sender.New(...)/AtAsync(...) and await
SendAsync() or async disposal to ensure outstanding frames are flushed.
In `@src/net-questdb-client-tests/Qwp/QwpBitWriterTests.cs`:
- Around line 83-93: Replace the manual try/catch in the test method
WriteBits_ExhaustedBuffer_Throws with NUnit's Assert.Throws pattern: call
Assert.Throws<InvalidOperationException>(() => w.WriteBits(0, 1)) to assert the
exception from QwpBitWriter.WriteBits and remove the thrown variable/try-catch
block so the test uses the standard idiomatic assertion used elsewhere in the
suite.
In `@src/net-questdb-client-tests/Qwp/QwpWebSocketSenderTests.cs`:
- Around line 259-265: The test ConnectFailure_ClosedPort_RaisesIngressError
currently assumes port 1 is closed; instead reserve an ephemeral local port and
use that to guarantee a closed/unused port: in the test (method
ConnectFailure_ClosedPort_RaisesIngressError) create a TcpListener bound to
IPAddress.Loopback with port 0 to obtain an available port, read the assigned
port, stop/close the listener to free that port, then construct the Sender.New
connection string using the obtained port (e.g.
"ws::addr=127.0.0.1:{port};auto_flush=off;") and assert the IngressError with
ErrorCode.SocketError as before.
In `@src/net-questdb-client/Qwp/QwpInFlightWindow.cs`:
- Around line 29-48: The class XML doc for QwpInFlightWindow is too verbose and
restates behavior; trim it to a short "why" note that highlights only
non-obvious constraints/decisions (e.g., sentinel choice of -1 for AckedSequence
and HighestSentSequence to disambiguate "never" vs sequence 0, the
cumulative-ACK semantics that AcknowledgeUpTo absorbs lower/out-of-order seqs
and throws on seq > HighestSentSequence as a server bug, and the
terminal-failure semantics where FailAll records the first failure rethrown by
AwaitEmpty). Remove procedural/descriptive narration about how the class works
and keep only these hidden constraints and counter-intuitive choices referenced
by their symbols (QwpInFlightWindow, AckedSequence, HighestSentSequence,
AcknowledgeUpTo, FailAll, AwaitEmpty).
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: 414a10c2-10ec-43b0-9208-72c8acb58d2b
📒 Files selected for processing (10)
README.mdsrc/net-questdb-client-tests/Qwp/QwpBitWriterTests.cssrc/net-questdb-client-tests/Qwp/QwpInFlightWindowTests.cssrc/net-questdb-client-tests/Qwp/QwpWebSocketSenderTests.cssrc/net-questdb-client/Qwp/QwpInFlightWindow.cssrc/net-questdb-client/Qwp/QwpWebSocketTransport.cssrc/net-questdb-client/Qwp/Sf/QwpCursorSendEngine.cssrc/net-questdb-client/Qwp/Sf/QwpMmapSegment.cssrc/net-questdb-client/Qwp/Sf/QwpSegmentRing.cssrc/net-questdb-client/Senders/QwpWebSocketSender.cs
🚧 Files skipped from review as they are similar to previous changes (2)
- src/net-questdb-client/Qwp/Sf/QwpMmapSegment.cs
- src/net-questdb-client/Senders/QwpWebSocketSender.cs
There was a problem hiding this comment.
🧹 Nitpick comments (1)
src/net-questdb-client/Senders/QwpWebSocketSender.cs (1)
103-107: Consider documenting the async-only requirement.The check correctly enforces
in_flight_window > 1, but the error message says "async mode" without clarifying that sync mode (in_flight_window=1) requires a different sender implementation or isn't supported for WebSocket transport. A brief comment or more descriptive error message would help users understand their options.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/net-questdb-client/Senders/QwpWebSocketSender.cs` around lines 103 - 107, Update the config error to be explicit and add a short comment: in the QwpWebSocketSender constructor (or where options.in_flight_window is validated) change the IngressError message to state that WebSocket transport requires asynchronous mode (in_flight_window > 1) and that synchronous mode (in_flight_window == 1) is unsupported for this sender and a different sender implementation must be used; also add a one-line comment above the check referencing options.in_flight_window and the requirement so future readers see the limitation clearly (keep using ErrorCode.ConfigError and IngressError as the thrown types).
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Nitpick comments:
In `@src/net-questdb-client/Senders/QwpWebSocketSender.cs`:
- Around line 103-107: Update the config error to be explicit and add a short
comment: in the QwpWebSocketSender constructor (or where
options.in_flight_window is validated) change the IngressError message to state
that WebSocket transport requires asynchronous mode (in_flight_window > 1) and
that synchronous mode (in_flight_window == 1) is unsupported for this sender and
a different sender implementation must be used; also add a one-line comment
above the check referencing options.in_flight_window and the requirement so
future readers see the limitation clearly (keep using ErrorCode.ConfigError and
IngressError as the thrown types).
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: 9cffd06e-584e-428b-8f6d-47bbd5db15ef
📒 Files selected for processing (6)
src/net-questdb-client-tests/Qwp/QwpInFlightWindowTests.cssrc/net-questdb-client/Qwp/QwpTableBuffer.cssrc/net-questdb-client/Qwp/QwpWebSocketTransport.cssrc/net-questdb-client/Qwp/Sf/QwpCursorSendEngine.cssrc/net-questdb-client/Senders/QwpWebSocketSender.cssrc/net-questdb-client/Utils/SenderOptions.cs
✅ Files skipped from review due to trivial changes (1)
- src/net-questdb-client/Qwp/QwpWebSocketTransport.cs
There was a problem hiding this comment.
Actionable comments posted: 4
♻️ Duplicate comments (1)
src/net-questdb-client/Qwp/Sf/QwpFiles.cs (1)
78-91:⚠️ Potential issue | 🟠 Major | ⚡ Quick winDon't fall back to “any
IOExceptionmeans locked”.
return ex.GetType() == typeof(IOException)still converts generic open failures intonull.QwpSlotLock.TryAcquire()andQwpOrphanScanner.ClaimOrphans()treatnullas “someone else holds the lock”, so real filesystem faults can be silently skipped instead of surfaced.On .NET 6-10, what exception types / HResult values does FileStream(path, FileMode.OpenOrCreate, FileAccess.ReadWrite, FileShare.None) throw for a true sharing violation on Windows, Linux, and macOS?🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/net-questdb-client/Qwp/Sf/QwpFiles.cs` around lines 78 - 91, IsSharingViolation currently treats any plain IOException as a sharing/lock case, which hides real filesystem errors; update IsSharingViolation to only return true for explicit sharing-violation indicators (e.g. the existing sharingViolationHResult check and any additional concrete HResults or exception types you add), remove the fallback "return ex.GetType() == typeof(IOException)" and instead return false for unknown IOExceptions so callers like QwpSlotLock.TryAcquire and QwpOrphanScanner.ClaimOrphans will surface real errors; if you need cross-platform coverage, add explicit platform-specific HResult/errno mappings (checked via ex.HResult or platform APIs) rather than a blanket IOException check.
🧹 Nitpick comments (2)
src/net-questdb-client-tests/Qwp/Sf/QwpMmapSegmentTests.cs (1)
189-212: ⚡ Quick winAssert post-reopen frame contents, not only envelope counts.
This test can still pass if envelope accounting is correct but replay/write placement is wrong. Verify both frames via
TryReadFrameafter the third reopen.Suggested test hardening
using var third = QwpMmapSegment.Open(path, 4096, 0); Assert.That(third.EnvelopeCount, Is.EqualTo(2)); + var dest = new byte[64]; + var offset = QwpMmapSegment.HeaderSize; + + var len1 = third.TryReadFrame(offset, dest, out var fsn1); + Assert.That(len1, Is.EqualTo(3)); + Assert.That(fsn1, Is.EqualTo(0)); + Assert.That(dest.AsSpan(0, len1).ToArray(), Is.EqualTo(new byte[] { 1, 2, 3 })); + + offset += QwpMmapSegment.EnvelopeHeaderSize + len1; + var len2 = third.TryReadFrame(offset, dest, out var fsn2); + Assert.That(len2, Is.EqualTo(3)); + Assert.That(fsn2, Is.EqualTo(1)); + Assert.That(dest.AsSpan(0, len2).ToArray(), Is.EqualTo(new byte[] { 99, 99, 99 }));🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/net-questdb-client-tests/Qwp/Sf/QwpMmapSegmentTests.cs` around lines 189 - 212, The test AppendAfterReopenWithCorruption_OverwritesTornBytes currently asserts only EnvelopeCount; extend it to validate frame contents by using QwpMmapSegment.TryReadFrame on the reopened segment (after the third Open) to read frame 0 and frame 1 and assert their payloads equal the original first append ([1,2,3]) and the subsequent append ([99,99,99]) respectively; use the QwpMmapSegment.TryReadFrame method (and any provided buffer/length results) to compare exact byte arrays rather than relying solely on EnvelopeCount.src/net-questdb-client/Utils/SenderOptions.cs (1)
360-389: 💤 Low value
token_x/token_ynot tracked byValidateWebSocketKeysAgainstDefaults.The
_*UserSetpattern is used to reject WS-only keys on non-WS protocols when set programmatically, buttoken_xandtoken_ydon't have corresponding_tokenXUserSet/_tokenYUserSettracking fields. While these properties are[Obsolete], the validation inconsistency meansnew SenderOptions { protocol = ProtocolType.http, token_x = "..." }silently accepts a WS-only auth key.If these keys should be WS-only (matching the connection-string validation intent), add tracking fields and checks here.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/net-questdb-client/Utils/SenderOptions.cs` around lines 360 - 389, ValidateWebSocketKeysAgainstDefaults currently misses token_x and token_y because there are no corresponding user-set trackers; add boolean fields (e.g., _tokenXUserSet and _tokenYUserSet) to SenderOptions and set them when the token_x and token_y properties are assigned, then add checks in ValidateWebSocketKeysAgainstDefaults that call Throw(nameof(token_x)) and Throw(nameof(token_y)) when those trackers are true (consistent with the other _*UserSet checks) so WS-only tokens are rejected for non-ws schemes.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@src/net-questdb-client/Qwp/Sf/QwpBackgroundDrainerPool.cs`:
- Around line 157-168: The current code in QwpBackgroundDrainerPool
force-disposes all entries from _liveLocks (leakedLocks) even when a drain
worker timed out and RunDrainAsync may still be executing
IQwpSlotDrainer.DrainAsync, allowing another sender to reacquire and replay the
same segment; change the shutdown logic so you only call SfCleanup.Dispose(l)
for leakedLocks after the worker has actually exited (i.e., after confirming the
drain task joined/completed), and on timeout do not clear or dispose _liveLocks
(leave them leaked to be cleaned by the task's finally or process exit); update
the branch that handles the timeout path to skip the lock release/clear and
ensure the successful shutdown path still performs the disposal and
_liveLocks.Clear().
In `@src/net-questdb-client/Qwp/Sf/QwpCursorSendEngine.cs`:
- Around line 417-429: The code currently ignores the result of loop.Wait and
proceeds to dispose shared resources; change it to capture whether the loop
actually joined (e.g., bool joined = loop is null ? true :
loop.Wait(TimeSpan.FromSeconds(5))) and only call
SfCleanup.Dispose(_segmentManager), SfCleanup.Dispose(_ring),
SfCleanup.Dispose(_slotLock) and UnlinkSegmentFiles(slotDir) if joined (or if
fullyDrained and joined as appropriate); leave cts?.Cancel() and
SfCleanup.Dispose(cts) as-is but do not tear down ring/lock/segment state when
joined is false so background send/receive won't run against disposed segments
or release the slot lock prematurely.
In `@src/net-questdb-client/Qwp/Sf/QwpSegmentManager.cs`:
- Around line 102-122: Dispose currently returns if
_workerTask.Wait(_shutdownWait) times out, allowing QwpCursorSendEngine.Dispose
to tear down _ring while ServiceRing may still be running; fix by ensuring
Dispose performs a hard join before disposing ring ownership: after calling
SfCleanup.Run(() => _cts.Cancel()) and SfCleanup.Run(() => _wakeup.Release()),
change the shutdown sequence in Dispose so that if _workerTask is not null you
first attempt the timed wait, and if that times out call _workerTask.Wait()
without a timeout (or otherwise block until completion) and only after the
worker task has completed call SfCleanup.Dispose(_cts) and
SfCleanup.Dispose(_wakeup) and allow QwpCursorSendEngine.Dispose to dispose the
ring—this guarantees ServiceRing has exited before _ring is torn down
(referencing Dispose, _workerTask, _shutdownWait, _cts, _wakeup, ServiceRing,
and QwpCursorSendEngine.Dispose).
In `@src/net-questdb-client/Utils/SenderOptions.cs`:
- Around line 913-918: The XML doc for the reconnect_max_backoff_millis property
is inaccurate (says "Defaults to 5 s") while the backing field
_reconnectMaxBackoff is initialized to 30 seconds; fix by making the
documentation consistent with the implementation: update the summary for
reconnect_max_backoff_millis to state "Defaults to 30 s" (or, if you intend a 5s
default, change the initializer of _reconnectMaxBackoff to
TimeSpan.FromMilliseconds(5000) and ensure _reconnectMaxBackoffUserSet behavior
remains unchanged), and run a quick search for any other doc/initializer
mismatches for reconnect backoff to keep them consistent.
---
Duplicate comments:
In `@src/net-questdb-client/Qwp/Sf/QwpFiles.cs`:
- Around line 78-91: IsSharingViolation currently treats any plain IOException
as a sharing/lock case, which hides real filesystem errors; update
IsSharingViolation to only return true for explicit sharing-violation indicators
(e.g. the existing sharingViolationHResult check and any additional concrete
HResults or exception types you add), remove the fallback "return ex.GetType()
== typeof(IOException)" and instead return false for unknown IOExceptions so
callers like QwpSlotLock.TryAcquire and QwpOrphanScanner.ClaimOrphans will
surface real errors; if you need cross-platform coverage, add explicit
platform-specific HResult/errno mappings (checked via ex.HResult or platform
APIs) rather than a blanket IOException check.
---
Nitpick comments:
In `@src/net-questdb-client-tests/Qwp/Sf/QwpMmapSegmentTests.cs`:
- Around line 189-212: The test
AppendAfterReopenWithCorruption_OverwritesTornBytes currently asserts only
EnvelopeCount; extend it to validate frame contents by using
QwpMmapSegment.TryReadFrame on the reopened segment (after the third Open) to
read frame 0 and frame 1 and assert their payloads equal the original first
append ([1,2,3]) and the subsequent append ([99,99,99]) respectively; use the
QwpMmapSegment.TryReadFrame method (and any provided buffer/length results) to
compare exact byte arrays rather than relying solely on EnvelopeCount.
In `@src/net-questdb-client/Utils/SenderOptions.cs`:
- Around line 360-389: ValidateWebSocketKeysAgainstDefaults currently misses
token_x and token_y because there are no corresponding user-set trackers; add
boolean fields (e.g., _tokenXUserSet and _tokenYUserSet) to SenderOptions and
set them when the token_x and token_y properties are assigned, then add checks
in ValidateWebSocketKeysAgainstDefaults that call Throw(nameof(token_x)) and
Throw(nameof(token_y)) when those trackers are true (consistent with the other
_*UserSet checks) so WS-only tokens are rejected for non-ws schemes.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: 0fd63c39-99d1-4f6d-9b63-76e39bebe1b4
📒 Files selected for processing (24)
CLAUDE.mdsrc/net-questdb-client-tests/Qwp/QwpColumnTests.cssrc/net-questdb-client-tests/Qwp/QwpSymbolDictionaryTests.cssrc/net-questdb-client-tests/Qwp/QwpWebSocketSenderTests.cssrc/net-questdb-client-tests/Qwp/QwpWebSocketTransportTests.cssrc/net-questdb-client-tests/Qwp/Sf/QwpBackgroundDrainerPoolTests.cssrc/net-questdb-client-tests/Qwp/Sf/QwpCrc32CTests.cssrc/net-questdb-client-tests/Qwp/Sf/QwpFilesTests.cssrc/net-questdb-client-tests/Qwp/Sf/QwpMmapSegmentTests.cssrc/net-questdb-client-tests/Qwp/Sf/QwpOrphanScannerTests.cssrc/net-questdb-client-tests/SenderOptionsTests.cssrc/net-questdb-client/Qwp/QwpColumn.cssrc/net-questdb-client/Qwp/QwpSymbolDictionary.cssrc/net-questdb-client/Qwp/QwpTableBuffer.cssrc/net-questdb-client/Qwp/QwpWebSocketTransport.cssrc/net-questdb-client/Qwp/Sf/QwpBackgroundDrainerPool.cssrc/net-questdb-client/Qwp/Sf/QwpCursorSendEngine.cssrc/net-questdb-client/Qwp/Sf/QwpFiles.cssrc/net-questdb-client/Qwp/Sf/QwpMmapSegment.cssrc/net-questdb-client/Qwp/Sf/QwpOrphanScanner.cssrc/net-questdb-client/Qwp/Sf/QwpSegmentManager.cssrc/net-questdb-client/Senders/ISender.cssrc/net-questdb-client/Senders/QwpWebSocketSender.cssrc/net-questdb-client/Utils/SenderOptions.cs
✅ Files skipped from review due to trivial changes (2)
- CLAUDE.md
- src/net-questdb-client-tests/Qwp/Sf/QwpFilesTests.cs
🚧 Files skipped from review as they are similar to previous changes (4)
- src/net-questdb-client-tests/Qwp/Sf/QwpOrphanScannerTests.cs
- src/net-questdb-client-tests/Qwp/Sf/QwpCrc32CTests.cs
- src/net-questdb-client/Qwp/Sf/QwpMmapSegment.cs
- src/net-questdb-client-tests/Qwp/Sf/QwpBackgroundDrainerPoolTests.cs
Captures 37 review passes (~140 findings) into a rationale file plus a compacted action plan. The action plan groups findings by severity (10 HIGH, ~30 MED, ~80 LOW), proposes a three-PR sequencing, and records the decisions consciously made (forward-compat behaviour, sender separation, error code divergence) to prevent re-litigation. The session pattern follow-up proposal stays separate for now. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Contributor
|
Some early investigation, please cross-check against your un-pushed changes!
|
feat(qwp): add QwpQueryClient for QWP egress over WebSocket
kafka1991
changed the title
Member
|
@kafka1991 — code review pass. Findings below were verified against the actual source (a longer initial pass produced ~25 BLOCKERs and HIGHs; most retracted on re-reading — e.g. the symbol-dict isn't actually leaky because HIGHSF: no fsync between append and ACK — SF: recovery scan is O(n) per segment — SF: orphan-scanner PID liveness is unreliable — SF: Egress: CREDIT not coalesced — Egress: array-dim multiplication overflow guard is brittle — Egress: symbol-dict varint range cast — MEDIUMAsync dispose race on
Connect-string
Decoder
LOW
Not bugs — verified and dismissed
This was an AI-assisted review (Claude Code, Opus 4.7). Each finding was verified against the source but please re-confirm before acting. Happy to drop concrete patches for any of the SF items if useful. |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
3 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Adds full QWP WebSocket support to the .NET client — pipelined ingest sender (
ws::/wss::), egress query client (QueryClient.New(...)), and opt-in store-and-forward (SF) for crash-safe ingest replay through transient outages. Functional parity with Java PR #11 (egress) and PR #17 (SF).What's new
Ingest (
ws::/wss::)Sender.New("ws::addr=host:9000;in_flight_window=128;...")over the QWP columnar binary protocol; pipelined send with bounded ACK window.IQwpWebSocketSenderinterface (Ping,GetHighestAckedSeqTxn,GetHighestDurableSeqTxn) — cast required for WS-specific methods.auto_flush_rows/_interval/_bytestriggers route through the same dispatch path as HTTP/TCP.sf_dir=...): outgoing batches mmap'd to disk before send; cumulative ACK trims sealed segments; transient failures replay silently.tls_verify=unsafe_offfor self-signed certs;token_x/token_yauth.503 + X-QuestDB-Role: REPLICA / PRIMARY_CATCHUProtated past automatically (full role enforcement requires server-side PR chore(qwp): advertise role on /write/v4 upgrade for primary failover questdb#7061).Egress (
QueryClient.New(...))Sender.New(QueryOptionsvsSenderOptionsstay disjoint).QwpResultBatchDecoderforRESULT_BATCH/RESULT_END/QUERY_ERROR/EXEC_DONE/CACHE_RESET/SERVER_INFO; per-column scratches grow-and-reuse across batches.QwpBindValues— typed bind builder with 18 wire types, ascending-index validation, scale/precision range checks; byte vectors pinned inQwpBindValuesVectorsTests.target=any|primary|replicafilter,QwpServerInfo/QwpRoleMismatchException,OnFailoverResethandler callback.compression=zstd|raw|auto,X-QWP-Accept-Encodingupgrade header, per-batchFLAG_ZSTD, streaming-mode fallback when batch size unknown.QueryOptions.InitialCredit+ per-batch CREDIT flow control.src/example-qwp-query/Program.cs(basic / binds / errors).Shared failover primitives
Egress and the SF ingress reconnect loop share
QwpHostHealthTracker(Healthy → Unknown → TransientReject → TransportError → TopologyReject priority),UniformDoubleJitter[base, 2·base)backoff, per-host connect timeout (auth_timeout_ms, 15 s default) bounding the WS upgrade, total wall-clock budget independent of attempt count, and mid-stream failure demotion so the next reconnect prefers an unattempted endpoint. Different budgets by design: 30 s for egress (interactive) vs 5 min for SF (background, has local mmap buffer).Performance
Benchmarks run in-process against a real QuestDB master build with
/write/v4+/read/v1enabled on127.0.0.1:9000(not against a mock server). Full data + methodology indocs/qwp-benchmarks.md.Ingest
Egress
/read/v1vs HTTP/execPer-column scratch reuse across batches gives ~1170× allocation reduction vs a naïve per-batch decoder (
BenchQueryWs).