diff --git a/CHANGES.md b/CHANGES.md index 74392012..eddc557f 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -22,6 +22,7 @@ This release makes the SDK stop misrepresenting what a node sent. It is a delibe | Dropped stream events | invisible | `client.DroppedStreamMessages` counts them; `StreamMessageQueueCapacity` sizes the queue | | Frames from a stale or retiring session | delivered as current | dropped, and counted by `Connection.StaleSessionFramesDropped` | | Frames dispatched outside the queue | invisible | counted by `Connection.FallbackDispatchedStreamMessages` | +| Frames answered while `OnConnected` runs | dispatched outside the queue, concurrently | queued like every other frame | | Stream event type | `LedgerStream.Type` was a public field | inherited `BaseStream.Type` property — source-compatible, but an assembly built against the old package needs a rebuild | | Stream event type value | `ResponseStreamType Type` | `ResponseStreamType? Type` — an event that carried no `type` no longer reports `UNKNOWN` as though the node had said so | @@ -41,6 +42,12 @@ The raw-response work, all five levels landing together. The problem: a consumer The entries below are grouped by what changed, not by the order the levels were built in. The short version of what will not compile is in the table above. +* **The stream queue now exists before consumer code can subscribe** (#113). `StartMessageProcessor` ran at the very end of `OnceOpen`, after `ResolveAllAwaiting()` and after the `OnConnected` callback. Subscribing from that callback is the ordinary pattern - it is what the wallet consuming this SDK does - and the node can answer the subscription before the handler returns, so the first frames found no channel and took the fallback: outside `StreamMessageQueueCapacity`, uncounted by `DroppedStreamMessages`, dispatched concurrently rather than one at a time. The events most likely to arrive out of order were precisely the first ones after connecting. + * **the blocker was an unrelated coupling, and that is the real fix.** `StartPingTimer` begins with `StopPingTimerSync`, which also called `StopMessageProcessor`, so simply starting the processor earlier had it torn down again moments later - measurably: `TestUDroppedStreamMessagesCountsWhatTheConsumerNeverSaw` went from 10 evictions to 0. Stopping a ping timer has no business stopping the message queue; the two lifecycles are now separate + * the processor is stopped explicitly where a connection genuinely ends - `Disconnect`, `DisconnectAndWaitAsync`, `OnceClose`, `OnConnectHandlerFailedAsync`, `ChangeServer` and `RetireCurrentSessionAndReconnectAsync` - rather than by side effect of a timer + * the fallback path stays, and that was measured rather than assumed: declaring it unreachable fails 6 tests. It still serves `OnMessage` on a client that never connected, anything arriving after the processor is stopped, and a write the channel refuses because its writer is already completed + * `FallbackDispatchedStreamMessages`, added alongside, is what makes this checkable from outside. It counts cumulatively and the fallback stays legitimate outside a connection, so the thing to watch is not an absolute zero but a zero *delta*: the counter should not move while a connection is being established + * **The model stopped inventing values it was never sent, and stopped losing ones it does not know** (**breaking**) — the second half of the raw-response work. Re-serializing a typed response used to differ from what arrived in both directions. Measured on live mainnet responses at `api_version = 2`: a ten-entry `account_tx` gained **156 fabricated members** and dropped 28. After this level the same capture gains **4** and drops 15, and every fabricated member left is the `Amount`/`DeliverMax` rename, which is the next level. * **Scope came from the protocol, not from a hand count.** The repository already vendors rippled's `ledger_entries.macro` and parses which fields are Required, Optional or Default. A new conformance test, `TestUNullabilityConformance`, reads it: a field the protocol allows to be absent must map to a property that can express absence. It found **9 direct violations** — `AMM.TradingFee`, `FeeSettings.ReferenceFeeUnits/ReserveBase/ReserveIncrement`, `LedgerHashes.FirstLedgerSequence/LastLedgerSequence`, `MPTokenIssuance.AssetScale`, `PayChannel.SourceTag/DestinationTag` * **A second, broader rule applies to the same models.** rippled's requirement flag describes the ledger object, but these models double as the contents of `PreviousFields`, `FinalFields` and `NewFields` — and `PreviousFields` carries only the members a transaction changed, so there even a Required field can be missing. Hence every value-typed property of a ledger-entry model is now nullable. Counted as the conformance test counts them — pairs of (model, property), which is what the defect is measured in — that is **80 pairs**; as a diff it is **50 property declarations across 23 files**, the difference being `LedgerEntryType` on the shared base, which one edit covers for all 31 models. On top of that, 21 properties on transaction models and `AccountInfo.LedgerIndex`/`LedgerCurrentIndex` diff --git a/Tests/Xrpl.Tests/Client/TestUStaleSessionFrames.cs b/Tests/Xrpl.Tests/Client/TestUStaleSessionFrames.cs index 110e5ac2..5d36f59f 100644 --- a/Tests/Xrpl.Tests/Client/TestUStaleSessionFrames.cs +++ b/Tests/Xrpl.Tests/Client/TestUStaleSessionFrames.cs @@ -85,9 +85,10 @@ private static async Task DriveWitnessAndWait(XrplClient client, ConcurrentQueue /// Waits until stream frames are queued rather than taking the fallback path. /// /// - /// Connect() returning is not enough: OnceOpen resolves the waiters first and - /// starts the message processor last, so a test that injects a frame the moment it connects - /// can find no queue to put it in. + /// Connect() returning implies this now - OnceOpen starts the processor before it + /// resolves the waiters - so the wait is a guard rather than a necessity. It earns its place by + /// failing here, with a sentence saying so, if that ordering ever regresses: without it these + /// tests would quietly exercise the fallback path instead and still pass. /// private static async Task WaitForMessageProcessor(XrplClient client) { diff --git a/Tests/Xrpl.Tests/Client/TestUStreamProcessorStartsBeforeCallbacks.cs b/Tests/Xrpl.Tests/Client/TestUStreamProcessorStartsBeforeCallbacks.cs new file mode 100644 index 00000000..17d715e9 --- /dev/null +++ b/Tests/Xrpl.Tests/Client/TestUStreamProcessorStartsBeforeCallbacks.cs @@ -0,0 +1,159 @@ +using Microsoft.VisualStudio.TestTools.UnitTesting; + +using System; +using System.Text; +using System.Threading; +using System.Threading.Tasks; + +using Xrpl.Client; +using Xrpl.Models.Subscriptions; +using Xrpl.Tests; + +namespace XrplTests.Client; + +/// +/// Frames answered while the OnConnected callback is still running go through the queue, +/// like every other frame. +/// +/// +/// Subscribing from OnConnected is the ordinary pattern - it is what the wallet consuming +/// this SDK does - and the node can answer that subscription before the handler returns. The +/// message processor used to start at the very end of OnceOpen, after the callback, so +/// those first frames found no channel and took the fallback: outside +/// StreamMessageQueueCapacity, uncounted by DroppedStreamMessages, and dispatched +/// concurrently rather than one at a time. The events most likely to arrive out of order were +/// precisely the first ones after connecting. +/// +/// Moving the start earlier was blocked by an unrelated coupling: StartPingTimer begins with +/// StopPingTimerSync, which stopped the message processor as well, so an earlier start was +/// torn down again moments later. The two lifecycles are now separate. +/// +/// +[TestClass] +public class TestUStreamProcessorStartsBeforeCallbacks +{ + private const string ScriptedReply = + "{\"id\":__ID__,\"status\":\"success\",\"type\":\"response\",\"result\":{}}"; + + private const string TransactionMessage = """ + { + "type": "transaction", + "status": "closed", + "validated": true, + "engine_result": "tesSUCCESS", + "tx_json": { "TransactionType": "Payment", "Account": "rP9jPyP5kyvFRb6ZiRghAGw5u8SGAmU4bd", "Sequence": 4 }, + "meta": { "AffectedNodes": [], "TransactionIndex": 0, "TransactionResult": "tesSUCCESS" } + } + """; + + /// + /// A frame driven from inside the OnConnected handler reaches the queue, not the + /// fallback. + /// + /// + /// FallbackDispatchedStreamMessages is the whole assertion: it counts frames dispatched + /// outside the queue, so before this change it would have counted this one. + /// + [TestMethod] + public async Task TestUFrameAnsweredDuringOnConnectedGoesThroughTheQueue() + { + using ScriptedResponseServer server = new ScriptedResponseServer(ScriptedReply); + using XrplClient client = new XrplClient(server.Url); + + bool channelExisted = false; + long fallbackDuringCallback = -1; + TaskCompletionSource callbackDone = new TaskCompletionSource( + TaskCreationOptions.RunContinuationsAsynchronously); + + client.connection.OnConnected += async () => + { + // Stands in for the node answering a subscription issued from this very handler. + channelExisted = client.connection.IsMessageProcessorRunning; + await client.connection.IOnMessageFastPath( + Encoding.UTF8.GetBytes(TransactionMessage), + client.connection.ActiveSessionId); + fallbackDuringCallback = client.connection.FallbackDispatchedStreamMessages; + callbackDone.TrySetResult(); + }; + + await client.Connect(); + + try + { + // Connect() returning is not enough: OnceOpen resolves the waiters before invoking + // this callback, so without the wait the assertions below can read their defaults. + Task finished = await Task.WhenAny(callbackDone.Task, Task.Delay(TimeSpan.FromSeconds(5))); + Assert.AreSame(callbackDone.Task, finished, "the OnConnected handler never ran"); + + Assert.IsTrue(channelExisted, + "the queue did not exist while consumer code was subscribing - frames answered there take the fallback"); + Assert.AreEqual(0L, fallbackDuringCallback, + "a frame answered during OnConnected went round the queue"); + } + finally + { + await client.Disconnect(); + } + } + + /// + /// The queue survives the ping timer starting, which happens after the callback. + /// + /// + /// This is the coupling that blocked the fix: StartPingTimer calls + /// StopPingTimerSync, and that used to stop the message processor too. Starting the + /// processor early is worth nothing if the ping timer tears it down a moment later, and the + /// symptom is invisible - the fallback keeps delivering frames, just without any of the queue's + /// guarantees. + /// + [TestMethod] + public async Task TestUStartingThePingTimerLeavesTheProcessorRunning() + { + using ScriptedResponseServer server = new ScriptedResponseServer(ScriptedReply); + using XrplClient client = new XrplClient( + server.Url, + new XrplClient.ClientOptions { UseCustomPing = true }); + await client.Connect(); + + try + { + // The ping timer starts at the very end of OnceOpen, after Connect() has returned. + // Asserting before it runs would pass whether or not it takes the processor down - + // the test has to wait for the thing that used to break it. + DateTime deadline = DateTime.UtcNow.AddSeconds(5); + while (!client.connection.IsPingTimerRunning && DateTime.UtcNow < deadline) + { + await Task.Delay(10); + } + + Assert.IsTrue(client.connection.IsPingTimerRunning, "the ping timer never started"); + + Assert.IsTrue(client.connection.IsMessageProcessorRunning, + "the ping timer took the message processor down with it"); + + long fallbackBefore = client.connection.FallbackDispatchedStreamMessages; + + TaskCompletionSource received = new TaskCompletionSource( + TaskCreationOptions.RunContinuationsAsynchronously); + client.OnTransaction += r => + { + received.TrySetResult(r); + return Task.CompletedTask; + }; + + await client.connection.IOnMessageFastPath( + Encoding.UTF8.GetBytes(TransactionMessage), + client.connection.ActiveSessionId); + + Task completed = await Task.WhenAny(received.Task, Task.Delay(TimeSpan.FromSeconds(5))); + Assert.AreSame(received.Task, completed, "the frame never reached the handler"); + + Assert.AreEqual(0L, client.connection.FallbackDispatchedStreamMessages - fallbackBefore, + "the frame took the fallback, so the processor was not running after all"); + } + finally + { + await client.Disconnect(); + } + } +} diff --git a/Xrpl/Client/connection.cs b/Xrpl/Client/connection.cs index 4902fee6..b9848bba 100644 --- a/Xrpl/Client/connection.cs +++ b/Xrpl/Client/connection.cs @@ -437,10 +437,11 @@ private static WebSocketClient CreateWebSocket(string url, ConnectionOptions con /// its writer is already completed - and none of them were visible from outside until this /// counter existed. /// - /// It makes the startup window measurable rather than merely argued about: a handler that - /// subscribes from OnConnected can be reached before - /// has run, and this says how often that happened and to - /// how many frames. + /// Cumulative over the life of the connection, and a non-zero value is not by itself a fault: + /// a client that was driven before it connected, or after it disconnected, legitimately has + /// frames here. What means something is an increase across a connect: the startup window this + /// counter was added to measure is closed, so the number should not move while a connection is + /// being established. /// /// public long FallbackDispatchedStreamMessages => Interlocked.Read(ref _fallbackDispatchedStreamMessages); @@ -482,19 +483,36 @@ internal long? ActiveSessionId /// rather than taking the fallback path. /// /// - /// Exists for tests. Connect() returning does not imply it: OnceOpen resolves the - /// waiters through connectionManager.ResolveAllAwaiting() and only then invokes the - /// OnConnected callback and starts the processor, so a caller can be connected and still - /// be ahead of the queue. That ordering is the startup window described on - /// . - /// - /// - /// Volatile.Read rather than a plain read or _messageProcessorLock: - /// the field is written under that lock, and holding it here would mean waiting out + /// Exists for tests. OnceOpen now starts the processor before it resolves the waiters + /// through connectionManager.ResolveAllAwaiting() and before the OnConnected + /// callback, so a returned Connect() does imply a queue - it did not until the ping + /// timer stopped taking the processor down with it. + /// + /// Volatile.Read rather than a plain read or _messageProcessorLock: the field is + /// written under that lock, and holding it here would mean waiting out /// StopMessageProcessorInternal, which blocks up to two seconds on the reader task. + /// /// internal bool IsMessageProcessorRunning => Volatile.Read(ref _streamMessageChannel) != null; + /// + /// Whether the ping timer is up. + /// + /// + /// Exists for tests, and for one question in particular: the ping timer starts at the very end + /// of OnceOpen, after Connect() has already returned, so a test that wants to + /// know the processor survived has to wait for it rather than + /// assume it. Without that wait such a test can pass by asserting too early - before the thing + /// it is testing has had a chance to go wrong. + /// + /// The signal is exact for that purpose: begins by calling + /// , which clears this field, and only assigns it afterwards. + /// Seeing it non-null therefore means the teardown step - the one that used to take the + /// message processor with it - is already behind us. + /// + /// + internal bool IsPingTimerRunning => Volatile.Read(ref _pingCts) != null; + /// /// Completes the stream channel's writer without clearing the channel, reproducing the state /// StopMessageProcessorInternal leaves behind for anyone who read @@ -675,8 +693,11 @@ public async Task ChangeServer( // 1. Quick state cleanup - stop reconnect loop StopReconnectLoop(); - // 2. Cancel ping timer (but don't wait yet) + // 2. Cancel ping timer and the message processor (but don't wait yet). Stopping the + // processor is explicit since StopPingTimerSync no longer does it as a side effect - this + // session's queue goes with the session. StopPingTimerSync(); + StopMessageProcessor(); // 3. Reject all pending requests BEFORE waiting for ping // This allows the ping handler to receive OperationCanceledException and exit quickly @@ -811,8 +832,10 @@ private async Task RetireCurrentSessionAndReconnectAsync(string reason) // New session is created immediately without waiting. // Callbacks check session ID to ignore retiring sessions. - // 4. Stop ping timer (but don't wait yet) + // 4. Stop ping timer and the message processor (but don't wait yet) - the queue belongs + // to the session being retired. StopPingTimerSync(); + StopMessageProcessor(); // 5. Reject all pending requests BEFORE waiting for ping // This allows the ping handler to receive OperationCanceledException and exit quickly @@ -1274,6 +1297,7 @@ public async Task Disconnect() ClearReconnectState(); // Clear all reconnect state on user disconnect StopPingTimerSync(); + StopMessageProcessor(); // Reject pending requests so ping handler can exit quickly requestManager.RejectAllWithCancellation(); @@ -1329,6 +1353,7 @@ public async Task DisconnectAndWaitAsync(TimeSpan timeout, CancellationToken can ClearReconnectState(); // Clear all reconnect state on user disconnect StopPingTimerSync(); + StopMessageProcessor(); // Reject pending requests so ping handler can exit quickly requestManager.RejectAllWithCancellation(); @@ -1997,6 +2022,13 @@ private async Task OnceOpen(WebSocketClient connectedSocket, long sessionId) connectedSocket.ResetIntentionalDisconnect(); + // Before ResolveAllAwaiting and before the OnConnected callback, because both hand control + // to consumer code that subscribes - and the node can answer that subscription while the + // handler is still running. Frames arriving with no channel take the fallback: outside the + // capacity, uncounted by DroppedStreamMessages and dispatched concurrently, so the first + // events after connecting were exactly the ones that could arrive out of order. + StartMessageProcessor(); + try { connectionManager.ResolveAllAwaiting(); @@ -2018,13 +2050,6 @@ private async Task OnceOpen(WebSocketClient connectedSocket, long sessionId) // Start ping timer AFTER connection is fully established and all callbacks completed // This is outside try/catch to ensure it always runs on successful connection StartPingTimer(); - - // After StartPingTimer, not before: StartPingTimer calls StopPingTimerSync, which stops - // the message processor too - starting the processor first would have it torn down again - // before a single frame arrived. That coupling is the reason this cannot simply move - // ahead of the OnConnected callback, where it belongs; see the note on - // EnqueueStreamMessage. - StartMessageProcessor(); } /// @@ -2092,6 +2117,7 @@ await errorHandler reconnect: BuildReconnectInfo(failures)); StopPingTimerSync(); + StopMessageProcessor(); requestManager.RejectAllWithCancellation(); await WaitForPingToFinishAsync(); @@ -2224,8 +2250,10 @@ private async Task OnceClose(int? code, string? description, WebSocketClient clo return; } - // Only stop ping timer for current socket + // Only for the current socket - and the message processor goes with it, this connection + // is over. StopPingTimerSync(); + StopMessageProcessor(); // Check if this is a network drop (FailureReason set by WebSocketClient) var isNetworkDrop = closingSocket.FailureReason == SocketFailureReason.NetworkDrop; @@ -2857,6 +2885,29 @@ private void StartPingTimer() } } + /// + /// Stops the ping timer, and nothing else. + /// + /// + /// It used to stop the message processor too, which tied two unrelated lifecycles together: + /// begins by calling this, so starting the processor before the + /// ping timer had it torn down again moments later - and that is what forced + /// to the very end of OnceOpen, after the + /// OnConnected callback, leaving every frame answered during that callback to the + /// fallback path. The processor is now stopped explicitly wherever a connection genuinely + /// ends: , , OnceClose, + /// OnConnectHandlerFailedAsync, and + /// RetireCurrentSessionAndReconnectAsync - the same six places the side effect used to + /// fire, so when the processor stops is unchanged; only the spurious stop inside + /// is gone. + /// + /// ReconnectLoopAsync retires a session without stopping the processor, and deliberately + /// so: it did not stop it before this change either, OnceClose has already run by the + /// time it retries, and tears down any leftover when the + /// next connection opens. Adding a stop there would discard frames still queued from before the + /// drop, on a path where nothing shows that is wanted. + /// + /// private void StopPingTimerSync() { var cts = _pingCts; @@ -2879,8 +2930,6 @@ private void StopPingTimerSync() wasmTimer?.Dispose(); cts?.Dispose(); - - StopMessageProcessor(); } /// @@ -3592,13 +3641,19 @@ private bool IsFromLiveSession(long? sessionId) /// and single-reader ordering hold on every target. /// /// - /// One window remains, and it is not platform-specific: - /// runs at the end of OnceOpen, after the OnConnected callback. A handler that - /// subscribes there can see frames answered before the channel exists, and those take the - /// fallback below - outside the capacity, the eviction count and the ordering. Moving the - /// start ahead of the callback is not a one-line change: calls - /// StopPingTimerSync, which stops the message processor as well, so an earlier start is - /// torn down again moments later. Untangling that is tracked separately. + /// The startup window is closed too: used to run at the + /// very end of OnceOpen, after the OnConnected callback, so a handler + /// subscribing there saw its first frames answered before the channel existed and they took + /// the fallback below - outside the capacity, the eviction count and the ordering. It now runs + /// before the callback. That was not a one-line change: begins + /// with StopPingTimerSync, which used to stop the message processor as well, so an + /// earlier start was torn down again moments later; see the remarks on + /// . + /// + /// + /// The fallback stays, because it is still reachable: on a + /// client that never connected, anything arriving after the processor is stopped, and a write + /// the channel refuses because its writer is already completed. /// /// private void EnqueueStreamMessage(byte[] frame, long? sessionId = null)