From 8790bb79dc96008b308e625d2d6a34a50e144c35 Mon Sep 17 00:00:00 2001 From: Aleksandr Platonenkov Date: Thu, 20 Aug 2026 10:20:55 -0300 Subject: [PATCH 01/16] fix(client)!: one stream path, browsers included Closes #110. Under WebAssembly EnqueueStreamMessage started one fire-and-forget task per frame instead of writing to the queue. Everything the queue provides was absent there: StreamMessageQueueCapacity did not apply, DroppedStreamMessages stayed at zero however far handlers fell behind, the backlog was bounded by nothing, and concurrent dispatch could hand handlers events out of the order the node sent them - the last one silent and worse than loss, since it leaves state wrong rather than incomplete. The bypass looked like a leftover rather than a decision: StartMessageProcessor says the queue exists for "true async support in WebAssembly single-threaded environment". Confirmed by measurement rather than by reading, in the Blazor demo against mainnet: through the queue 1 004 transactions / 52 s 19.2 tx/s 13 ledgers on the bypass 462 transactions / 33 s 13.9 tx/s 8 ledgers No console errors, timestamps in order. The background reader is scheduled fine under WASM, so the browser branch is gone and there is one path on every target. The caveats added a commit ago - on DroppedStreamMessages, on the capacity option, on the interface member and in CHANGES - described a limitation that no longer exists, so they are removed rather than left to mislead. --- CHANGES.md | 2 +- Xrpl/Client/IXrplClient.cs | 4 ---- Xrpl/Client/connection.cs | 43 ++++++++++++++++++-------------------- 3 files changed, 21 insertions(+), 28 deletions(-) diff --git a/CHANGES.md b/CHANGES.md index 2be04e8a..c650624c 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -203,7 +203,7 @@ The entries below are grouped by what changed, not by the order the levels were * `Connection.DroppedStreamMessages` and `IXrplClient.DroppedStreamMessages` count the discards, across reconnects and `ChangeServer` alike, since one `Connection` serves them all. Any increase means events arrived and never reached a handler * `ConnectionOptions.StreamMessageQueueCapacity` (default 10 000, unchanged) sizes the queue - raise it for a consumer that must not miss events and can hold the frames, lower it to bound memory harder * the counter is incremented from `Channel`'s `itemDropped` callback, which runs inside `TryWrite` on the receive loop. It does nothing but increment for that reason: raising an event or logging there would put consumer code back on the path this queue exists to keep it off - the very failure #105 described but which the queue already prevented - * **none of this applies under WebAssembly**, where `EnqueueStreamMessage` dispatches each frame directly instead of queueing it: the capacity is not consulted, nothing is evicted, and the counter stays at zero however far handlers fall behind - the backlog there is bounded by memory alone. Routing browser frames through the queue needs verifying in a real browser, which no test in this repository can do, so it is left as its own change rather than done blind + * **browsers no longer take a separate path** (#110). `EnqueueStreamMessage` used to start one fire-and-forget task per frame under WebAssembly, bypassing the queue: the capacity was not consulted, nothing was evicted, the counter stayed at zero however far handlers fell behind, the backlog was bounded by nothing, and concurrent dispatch could hand handlers events out of the order the node sent them. The queue was built for that environment to begin with - `StartMessageProcessor` says "true async support in WebAssembly single-threaded environment" - and measurement confirms it works there. Running the Blazor demo against mainnet: **1 004 transactions over 52 s (19.2 tx/s, 13 ledgers)** through the queue, no console errors, timestamps in order, against **462 over 33 s (13.9 tx/s)** on the bypass. One path now, so capacity, eviction counting and single-reader ordering hold on every target * the issue's premise was checked and does not hold: handlers do not run in the receive loop. `ProcessStreamMessageAsync` is called by a background reader, and the receive loop only calls `TryWrite`. What is real is the silent loss, which is what this addresses ## 10.12.0.0 08/16/2026 diff --git a/Xrpl/Client/IXrplClient.cs b/Xrpl/Client/IXrplClient.cs index ce55ec34..19609135 100644 --- a/Xrpl/Client/IXrplClient.cs +++ b/Xrpl/Client/IXrplClient.cs @@ -66,10 +66,6 @@ public interface IXrplClient : IDisposable /// and drops the oldest when full, so a slow handler costs events instead of stalling the /// socket - silently, until something reads this. A consumer building state from the /// stream should treat any increase as a signal that its state has drifted. - /// - /// Stays at zero under WebAssembly, where frames bypass the queue entirely - see - /// . - /// /// long DroppedStreamMessages { get; } diff --git a/Xrpl/Client/connection.cs b/Xrpl/Client/connection.cs index eb6d24e6..d0ec793e 100644 --- a/Xrpl/Client/connection.cs +++ b/Xrpl/Client/connection.cs @@ -268,10 +268,6 @@ public class ConnectionOptions /// Raise it for a consumer that must not miss events and can absorb the memory (each slot /// holds one frame); lower it to bound memory harder, accepting more loss. Default 10 000. /// - /// - /// Has no effect under WebAssembly, where frames are dispatched directly rather - /// than queued - see . - /// /// public int StreamMessageQueueCapacity { get; set; } = 10000; } @@ -440,14 +436,6 @@ private static WebSocketClient CreateWebSocket(string url, ConnectionOptions con /// Counts across the lifetime of this connection, including across reconnects and /// ChangeServer, since the same object serves them all. /// - /// - /// Not counted under WebAssembly. In a browser EnqueueStreamMessage dispatches - /// each frame directly instead of queueing it, so - /// does not apply, nothing is ever - /// evicted, and this stays at zero however far handlers fall behind. The backlog there is - /// bounded by nothing but memory. Routing browser frames through the queue is a separate - /// change - it needs verifying in a real browser, which no test here can do. - /// /// public long DroppedStreamMessages => Interlocked.Read(ref _droppedStreamMessages); private CancellationTokenSource? _messageProcessorCts = null; @@ -3405,25 +3393,34 @@ string Text() } /// - /// Routes a message to stream processing (Channel or fire-and-forget).
- /// Used for both regular stream messages (no "id") and async follow-ups - /// that have "id" but no matching pending request (e.g. path_find updates). + /// Hands a stream message to the background processor. /// + /// + /// Used for ordinary stream messages (no id) and for follow-ups carrying an id + /// that matches no pending request, such as path_find updates. + /// + /// Browsers used to take a separate path here - one fire-and-forget task per frame, bypassing + /// the queue entirely, so did not + /// apply, stayed at zero however far handlers fell behind, + /// the backlog was bounded by nothing, and concurrent dispatch could hand handlers events out + /// of the order the node sent them. The queue was built for this environment in the first + /// place ("true async support in WebAssembly single-threaded environment" on + /// ), and measurement confirmed it works there: running the + /// Blazor demo against mainnet, the queue delivered 1 004 transactions over 52 s (19.2 tx/s, + /// 13 ledgers) with no console errors and timestamps in order - against 462 over 33 s + /// (13.9 tx/s) on the bypass. So there is one path now, and capacity, eviction counting and + /// single-reader ordering hold everywhere. + /// + /// private void EnqueueStreamMessage(byte[] frame) { - if (OperatingSystem.IsBrowser()) - { - _ = ProcessStreamMessageFireAndForgetAsync(frame); - } - else { var channel = _streamMessageChannel; if (channel != null) { // No failure branch on purpose: the channel is bounded with DropOldest, so - // TryWrite always succeeds and silently evicts the oldest frame instead. A - // consumer falling 10 000 messages behind loses the oldest events with nothing - // reported - worth knowing, but a log line here would never fire. + // TryWrite always succeeds and silently evicts the oldest frame instead - counted + // by the itemDropped callback rather than reported here. channel.Writer.TryWrite(frame); } else From c9fb4c41085c077ada1cfff209a76f7119acf12e Mon Sep 17 00:00:00 2001 From: Aleksandr Platonenkov Date: Thu, 20 Aug 2026 11:07:19 -0300 Subject: [PATCH 02/16] docs(client): say what the single stream path does and does not cover Review pointed out the claim was too broad: with the browser bypass gone the platforms match, but frames can still miss the queue - StartMessageProcessor runs at the end of OnceOpen, after the OnConnected callback, and a handler that subscribes there can see answers before the channel exists. Those take the fallback, outside the capacity, the eviction count and the ordering. Moving the start ahead of the callback was tried and reverted, because it makes things worse in a way worth recording: StartPingTimer calls StopPingTimerSync, which stops the message processor as well, so the earlier start is torn down moments later and every frame takes the fallback from then on. The eviction test drops from 10 to 0. Stopping a ping timer should not stop the message queue, and untangling that is its own change - #113. So the wording now states the window rather than claiming it away, and the ordering constraint is written down at both StartMessageProcessor and EnqueueStreamMessage so the next person does not rediscover it by breaking a test. --- CHANGES.md | 2 +- Xrpl/Client/connection.cs | 21 +++++++++++++++++---- 2 files changed, 18 insertions(+), 5 deletions(-) diff --git a/CHANGES.md b/CHANGES.md index c650624c..eebf9a39 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -203,7 +203,7 @@ The entries below are grouped by what changed, not by the order the levels were * `Connection.DroppedStreamMessages` and `IXrplClient.DroppedStreamMessages` count the discards, across reconnects and `ChangeServer` alike, since one `Connection` serves them all. Any increase means events arrived and never reached a handler * `ConnectionOptions.StreamMessageQueueCapacity` (default 10 000, unchanged) sizes the queue - raise it for a consumer that must not miss events and can hold the frames, lower it to bound memory harder * the counter is incremented from `Channel`'s `itemDropped` callback, which runs inside `TryWrite` on the receive loop. It does nothing but increment for that reason: raising an event or logging there would put consumer code back on the path this queue exists to keep it off - the very failure #105 described but which the queue already prevented - * **browsers no longer take a separate path** (#110). `EnqueueStreamMessage` used to start one fire-and-forget task per frame under WebAssembly, bypassing the queue: the capacity was not consulted, nothing was evicted, the counter stayed at zero however far handlers fell behind, the backlog was bounded by nothing, and concurrent dispatch could hand handlers events out of the order the node sent them. The queue was built for that environment to begin with - `StartMessageProcessor` says "true async support in WebAssembly single-threaded environment" - and measurement confirms it works there. Running the Blazor demo against mainnet: **1 004 transactions over 52 s (19.2 tx/s, 13 ledgers)** through the queue, no console errors, timestamps in order, against **462 over 33 s (13.9 tx/s)** on the bypass. One path now, so capacity, eviction counting and single-reader ordering hold on every target + * **browsers no longer take a separate path** (#110). `EnqueueStreamMessage` used to start one fire-and-forget task per frame under WebAssembly, bypassing the queue: the capacity was not consulted, nothing was evicted, the counter stayed at zero however far handlers fell behind, the backlog was bounded by nothing, and concurrent dispatch could hand handlers events out of the order the node sent them. The queue was built for that environment to begin with - `StartMessageProcessor` says "true async support in WebAssembly single-threaded environment" - and measurement confirms it works there. Running the Blazor demo against mainnet: **1 004 transactions over 52 s (19.2 tx/s, 13 ledgers)** through the queue, no console errors, timestamps in order, against **462 over 33 s (13.9 tx/s)** on the bypass. The platforms no longer diverge: capacity, eviction counting and single-reader ordering hold on every target. One window remains and is not platform-specific - StartMessageProcessor runs after the OnConnected callback, so a handler subscribing there can see frames before the channel exists, and those take the direct fallback. Moving the start earlier is blocked by StartPingTimer calling StopPingTimerSync, which stops the processor too; tracked separately * the issue's premise was checked and does not hold: handlers do not run in the receive loop. `ProcessStreamMessageAsync` is called by a background reader, and the receive loop only calls `TryWrite`. What is real is the silent loss, which is what this addresses ## 10.12.0.0 08/16/2026 diff --git a/Xrpl/Client/connection.cs b/Xrpl/Client/connection.cs index d0ec793e..b96574e1 100644 --- a/Xrpl/Client/connection.cs +++ b/Xrpl/Client/connection.cs @@ -1901,8 +1901,12 @@ 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(); - - // Start background message processor for stream messages + + // 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(); } @@ -3408,8 +3412,17 @@ string Text() /// ), and measurement confirmed it works there: running the /// Blazor demo against mainnet, the queue delivered 1 004 transactions over 52 s (19.2 tx/s, /// 13 ledgers) with no console errors and timestamps in order - against 462 over 33 s - /// (13.9 tx/s) on the bypass. So there is one path now, and capacity, eviction counting and - /// single-reader ordering hold everywhere. + /// (13.9 tx/s) on the bypass. So the platforms no longer diverge: capacity, eviction counting + /// 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. /// /// private void EnqueueStreamMessage(byte[] frame) From e67c98d44d9767a74a7ed307eb810f20b5db77b1 Mon Sep 17 00:00:00 2001 From: Aleksandr Platonenkov Date: Thu, 20 Aug 2026 11:19:45 -0300 Subject: [PATCH 03/16] =?UTF-8?q?fix(client):=20=D0=BA=D0=B0=D0=B4=D1=80?= =?UTF-8?q?=D1=8B=20=D1=83=D1=85=D0=BE=D0=B4=D1=8F=D1=89=D0=B5=D0=B3=D0=BE?= =?UTF-8?q?=20=D1=81=D0=BE=D0=BA=D0=B5=D1=82=D0=B0=20=D0=B1=D0=BE=D0=BB?= =?UTF-8?q?=D1=8C=D1=88=D0=B5=20=D0=BD=D0=B5=20=D0=B2=D1=8B=D0=B4=D0=B0?= =?UTF-8?q?=D1=8E=D1=82=D1=81=D1=8F=20=D0=B7=D0=B0=20=D1=82=D0=B5=D0=BA?= =?UTF-8?q?=D1=83=D1=89=D0=B8=D0=B5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Идентичность сессии была протянута во все колбэки жизненного цикла - OnceOpen, OnConnectionFailed, OnceClose сверяются с _activeSession.SessionId - и путь сообщений оставался единственным исключением: ws.OnBinaryMessage звал IOnMessageFastPath(m) вообще без сессии. Это важно потому, что уход сокета не мгновенный. RetireOldSessionAsync работает fire-and-forget рядом с новым соединением и закрывает старый сокет *штатно*, то есть тот продолжает доставлять всё время рукопожатия закрытия - с живым колбэком. Всё, что он присылал, попадало в очередь новой сессии. После переподключения такие кадры просто устаревшие. После ChangeServer между сетями - хуже: переключились с mainnet на testnet, а обработчик, считающий что он на testnet, получает хвост потока mainnet: транзакции по счетам, которых там нет, индексы леджеров из другой цепи, и ничто не помечает их как принадлежащие прошлому соединению. Теперь сессия едет вместе с кадром и сверяется перед записью в очередь. null-сессия означает, что вызывающему нечего назвать (OnMessage, доступный кому угодно), и в этом случае не отбрасывается ничего. Connection.StaleSessionFramesDropped считает отброшенное - отдельно от DroppedStreamMessages, потому что смысл разный: ненулевое значение здесь нормально сразу после переподключения, а там означает, что потребители не успевают. Тест проверен мутацией: со снятой сверкой сессии кадр доходит до обработчика и тест краснеет. Closes #112 --- CHANGES.md | 6 + .../Client/TestUStaleSessionFrames.cs | 119 ++++++++++++++++++ Xrpl/Client/connection.cs | 61 +++++++-- 3 files changed, 178 insertions(+), 8 deletions(-) create mode 100644 Tests/Xrpl.Tests/Client/TestUStaleSessionFrames.cs diff --git a/CHANGES.md b/CHANGES.md index eebf9a39..eb1a6d7e 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -20,6 +20,7 @@ This release makes the SDK stop misrepresenting what a node sent. It is a delibe | Stream events | `client.connection.OnTransaction += …` | also `client.OnTransaction += …` — on `IXrplClient` now; the old form still works | | `IXrplClient.connection` | `{ get; set; }` | `{ get; }` — assigning it would strand handlers on the old object | | Dropped stream events | invisible | `client.DroppedStreamMessages` counts them; `StreamMessageQueueCapacity` sizes the queue | +| Frames from a retiring socket | delivered as current | dropped, and counted by `Connection.StaleSessionFramesDropped` | | 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 | @@ -205,6 +206,11 @@ The entries below are grouped by what changed, not by the order the levels were * the counter is incremented from `Channel`'s `itemDropped` callback, which runs inside `TryWrite` on the receive loop. It does nothing but increment for that reason: raising an event or logging there would put consumer code back on the path this queue exists to keep it off - the very failure #105 described but which the queue already prevented * **browsers no longer take a separate path** (#110). `EnqueueStreamMessage` used to start one fire-and-forget task per frame under WebAssembly, bypassing the queue: the capacity was not consulted, nothing was evicted, the counter stayed at zero however far handlers fell behind, the backlog was bounded by nothing, and concurrent dispatch could hand handlers events out of the order the node sent them. The queue was built for that environment to begin with - `StartMessageProcessor` says "true async support in WebAssembly single-threaded environment" - and measurement confirms it works there. Running the Blazor demo against mainnet: **1 004 transactions over 52 s (19.2 tx/s, 13 ledgers)** through the queue, no console errors, timestamps in order, against **462 over 33 s (13.9 tx/s)** on the bypass. The platforms no longer diverge: capacity, eviction counting and single-reader ordering hold on every target. One window remains and is not platform-specific - StartMessageProcessor runs after the OnConnected callback, so a handler subscribing there can see frames before the channel exists, and those take the direct fallback. Moving the start earlier is blocked by StartPingTimer calling StopPingTimerSync, which stops the processor too; tracked separately * the issue's premise was checked and does not hold: handlers do not run in the receive loop. `ProcessStreamMessageAsync` is called by a background reader, and the receive loop only calls `TryWrite`. What is real is the silent loss, which is what this addresses +* **Frames from a socket being retired no longer reach handlers as if they were current** (#112). Session identity was threaded into every lifecycle callback - `OnceOpen`, `OnConnectionFailed`, `OnceClose` all compare against `_activeSession.SessionId` - and the message path was the one exception: `ws.OnBinaryMessage` called `IOnMessageFastPath(m)` with no session at all. + * that mattered because retirement is not instant. `RetireOldSessionAsync` runs fire-and-forget beside the new connection and closes the old socket *gracefully*, so it keeps delivering for the length of the close handshake with its callback still attached. Whatever it sent landed in the new session's queue + * after a reconnect those frames are stale. After a `ChangeServer` between networks they are worse than stale: switching mainnet to testnet, a handler that believes it is on testnet could receive the tail of mainnet's stream - transactions for accounts that do not exist there, ledger indexes from another chain, nothing marking them as belonging to the previous connection + * the session now travels with the frame and is checked before the write. A `null` session means the caller has none to name (`OnMessage`, which anyone may call), and nothing is rejected in that case + * `Connection.StaleSessionFramesDropped` counts what was discarded, kept separate from `DroppedStreamMessages` because the two mean different things: a non-zero value here is normal right after a reconnect, while the other means consumers are falling behind ## 10.12.0.0 08/16/2026 diff --git a/Tests/Xrpl.Tests/Client/TestUStaleSessionFrames.cs b/Tests/Xrpl.Tests/Client/TestUStaleSessionFrames.cs new file mode 100644 index 00000000..bca7dd0e --- /dev/null +++ b/Tests/Xrpl.Tests/Client/TestUStaleSessionFrames.cs @@ -0,0 +1,119 @@ +using Microsoft.VisualStudio.TestTools.UnitTesting; + +using System; +using System.Text; +using System.Threading.Tasks; + +using Xrpl.Client; +using Xrpl.Models.Subscriptions; +using Xrpl.Tests; + +namespace XrplTests.Client; + +/// +/// A frame from a socket that is being retired must not reach handlers as if it were current. +/// +/// +/// Retirement is not instant: RetireOldSessionAsync runs fire-and-forget alongside the new +/// connection and closes the old socket gracefully, so that socket keeps delivering for as long as +/// the close handshake takes - with its message callback still attached. After a reconnect those +/// frames are merely stale. After a ChangeServer between networks they describe a different +/// chain: transactions for accounts that do not exist on the new one, ledger indexes from +/// somewhere else entirely. +/// +/// The lifecycle callbacks have always compared their captured session against the active one; +/// the message path was the exception, carrying no session at all. +/// +/// +[TestClass] +public class TestUStaleSessionFrames +{ + 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": 9 }, + "meta": { "AffectedNodes": [], "TransactionIndex": 0, "TransactionResult": "tesSUCCESS" } + } + """; + + /// + /// A frame tagged with a session that is not the active one is dropped and counted. + /// + [TestMethod] + public async Task TestUFrameFromARetiredSessionNeverReachesHandlers() + { + using ScriptedResponseServer server = new ScriptedResponseServer(ScriptedReply); + using XrplClient client = new XrplClient(server.Url); + await client.Connect(); + + int calls = 0; + client.OnTransaction += _ => + { + calls++; + return Task.CompletedTask; + }; + + try + { + // long.MaxValue stands in for a session id that is no longer active - sessions are + // numbered upward from the first connection, so this can never be the live one. + await client.connection.IOnMessageFastPath( + Encoding.UTF8.GetBytes(TransactionMessage), + sessionId: long.MaxValue); + + Assert.AreEqual(1L, client.connection.StaleSessionFramesDropped, + "a frame from a session that is not active must be dropped before it reaches the queue"); + Assert.AreEqual(0, calls, "the handler saw a frame belonging to a connection that is being retired"); + } + finally + { + await client.Disconnect(); + } + } + + /// + /// The guard must not reject frames from the live session, or the stream stops entirely. + /// + /// + /// Without this, a comparison that always failed would pass the test above while breaking + /// every subscription in the SDK. + /// + [TestMethod] + public async Task TestUFrameFromTheActiveSessionIsDelivered() + { + using ScriptedResponseServer server = new ScriptedResponseServer(ScriptedReply); + using XrplClient client = new XrplClient(server.Url); + await client.Connect(); + + TaskCompletionSource received = new TaskCompletionSource( + TaskCreationOptions.RunContinuationsAsynchronously); + client.OnTransaction += r => + { + received.TrySetResult(r); + return Task.CompletedTask; + }; + + try + { + // No session named: the ordinary OnMessage entry point, which anyone may call and + // which has no session to compare against. It must be delivered, not dropped. + await client.connection.OnMessage(TransactionMessage); + + Task completed = await Task.WhenAny(received.Task, Task.Delay(TimeSpan.FromSeconds(5))); + Assert.AreSame(received.Task, completed, "a frame with no session attached was dropped"); + + Assert.AreEqual(0L, client.connection.StaleSessionFramesDropped, + "nothing here came from a retired session"); + } + finally + { + await client.Disconnect(); + } + } +} diff --git a/Xrpl/Client/connection.cs b/Xrpl/Client/connection.cs index b96574e1..487a0880 100644 --- a/Xrpl/Client/connection.cs +++ b/Xrpl/Client/connection.cs @@ -422,6 +422,22 @@ private static WebSocketClient CreateWebSocket(string url, ConnectionOptions con private long _droppedStreamMessages; + private long _staleSessionFramesDropped; + + /// + /// How many stream frames were discarded because they came from a session that is no longer + /// active. + /// + /// + /// A socket being retired keeps delivering until its graceful close finishes, so a handful of + /// frames can arrive after a reconnect or ChangeServer has already moved on. They are + /// dropped rather than delivered: after a change of network they would otherwise describe a + /// different chain entirely. Counted separately from + /// , which is about consumers falling behind - these two + /// mean different things and a non-zero value here is normal right after a reconnect. + /// + public long StaleSessionFramesDropped => Interlocked.Read(ref _staleSessionFramesDropped); + /// /// How many stream messages have been discarded because the consumer fell behind. /// @@ -1106,8 +1122,11 @@ await errorHandler.Invoke( try { // Use fast-path processing to prioritize ping/pong responses - // and prevent head-of-line blocking from high-volume stream data - await IOnMessageFastPath(m); + // and prevent head-of-line blocking from high-volume stream data. + // The session travels with the frame so a late arrival from a socket being + // retired can be told apart from one on the live connection - see + // EnqueueStreamMessage. + await IOnMessageFastPath(m, capturedSession.SessionId); } catch (Exception ex) { @@ -3238,7 +3257,7 @@ private async Task ProcessStreamMessageAsync(byte[] frame) /// private Task IOnMessageFastPath(string message) { - return IOnMessageFastPath(message, null); + return IOnMessageFastPath(message, null, sessionId: null); } /// @@ -3254,7 +3273,15 @@ private Task IOnMessageFastPath(string message) /// internal Task IOnMessageFastPath(byte[] utf8Message) { - return IOnMessageFastPath(null, utf8Message); + return IOnMessageFastPath(null, utf8Message, sessionId: null); + } + + /// + /// As above, for a frame whose originating session is known. + /// + internal Task IOnMessageFastPath(byte[] utf8Message, long? sessionId) + { + return IOnMessageFastPath(null, utf8Message, sessionId); } /// @@ -3276,7 +3303,11 @@ internal Task IOnMessageFastPath(byte[] utf8Message) /// bytes the socket produced. Only the warning and error callbacks, which still take a string, /// ask for text at all, through Text(), and materialize it once and only then. /// - private async Task IOnMessageFastPath(string message, byte[] utf8Message) + /// + /// The session whose socket produced this frame, or when the caller has + /// no session to name - , which anyone may call directly. + /// + private async Task IOnMessageFastPath(string message, byte[] utf8Message, long? sessionId) { lastActivityTime = DateTime.UtcNow; @@ -3357,7 +3388,7 @@ string Text() { // Message has "id" but no matching pending request — this is an async // follow-up (e.g. path_find updates). Route to stream processing. - EnqueueStreamMessage(Frame()); + EnqueueStreamMessage(Frame(), sessionId); return; } @@ -3392,7 +3423,7 @@ string Text() { // This is a stream message (no "id") - process asynchronously // to avoid blocking the receive loop and causing ping timeouts - EnqueueStreamMessage(Frame()); + EnqueueStreamMessage(Frame(), sessionId); } } @@ -3425,8 +3456,22 @@ string Text() /// torn down again moments later. Untangling that is tracked separately. /// /// - private void EnqueueStreamMessage(byte[] frame) + private void EnqueueStreamMessage(byte[] frame, long? sessionId = null) { + // A retiring socket keeps delivering while InitiateGracefulCloseAsync completes, and that + // close runs fire-and-forget alongside the new connection. Without this check its last + // frames land in the new session's queue and reach handlers as if they were current - + // stale after a reconnect, and from an entirely different chain after a ChangeServer + // between networks. Lifecycle callbacks already guard the same way against _activeSession. + // + // A null sessionId means the caller cannot name a session (OnMessage, which anyone may + // call): nothing to compare, so nothing is rejected. + if (sessionId is not null && _activeSession?.SessionId != sessionId) + { + Interlocked.Increment(ref _staleSessionFramesDropped); + return; + } + { var channel = _streamMessageChannel; if (channel != null) From e2edea74de7bc7fb9d61071465e530a797350c2a Mon Sep 17 00:00:00 2001 From: Aleksandr Platonenkov Date: Thu, 20 Aug 2026 11:32:57 -0300 Subject: [PATCH 04/16] =?UTF-8?q?fix(client):=20=D1=81=D0=BE=D0=B2=D0=BF?= =?UTF-8?q?=D0=B0=D0=B4=D0=B5=D0=BD=D0=B8=D1=8F=20id=20=D0=BC=D0=B0=D0=BB?= =?UTF-8?q?=D0=BE=20=E2=80=94=20=D0=BA=D0=B0=D0=B4=D1=80=20=D1=83=D1=85?= =?UTF-8?q?=D0=BE=D0=B4=D1=8F=D1=89=D0=B5=D0=B9=20=D1=81=D0=B5=D1=81=D1=81?= =?UTF-8?q?=D0=B8=D0=B8=20=D1=82=D0=BE=D0=B6=D0=B5=20=D0=BE=D1=82=D0=B1?= =?UTF-8?q?=D1=80=D0=B0=D1=81=D1=8B=D0=B2=D0=B0=D0=B5=D1=82=D1=81=D1=8F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Сверка только по SessionId пропускала тот самый случай, ради которого всё и делалось. ChangeServer и цикл переподключения зовут MarkAsRetiring() на сессии, пока она ещё _activeSession: замену ставит только ConnectInternalAsync, позже. Кадры, пришедшие в этом окне, несут id ровно той сессии, которую сейчас хоронят, — и проходили сверку. Теперь флаг retiring входит в проверку, и проверка идёт под _sessionLock — так же, как в OnceOpen и остальных колбэках жизненного цикла. Замок здесь не формальность: IsRetiring — обычный bool, и публикуется он только через _sessionLock, так что чтение без замка не гарантирует увидеть запись вообще. Чтобы окно было воспроизводимо в тесте, добавлен internal-шов MarkActiveSessionRetiringForTests(): помечает активную сессию ровно так, как это делают оба продакшн-пути, и возвращает её id. В продакшне это окно достижимо только гонкой с реальным переподключением, а id хоронимой сессии иначе не наблюдаем. Тест проверен мутацией: со снятой клаузулой !IsRetiring краснеет ровно новый тест, два прежних остаются зелёными. --- CHANGES.md | 3 +- .../Client/TestUStaleSessionFrames.cs | 43 +++++++++++++++++ Xrpl/Client/connection.cs | 46 +++++++++++++++++-- 3 files changed, 87 insertions(+), 5 deletions(-) diff --git a/CHANGES.md b/CHANGES.md index eb1a6d7e..8f80e3d9 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -209,7 +209,8 @@ The entries below are grouped by what changed, not by the order the levels were * **Frames from a socket being retired no longer reach handlers as if they were current** (#112). Session identity was threaded into every lifecycle callback - `OnceOpen`, `OnConnectionFailed`, `OnceClose` all compare against `_activeSession.SessionId` - and the message path was the one exception: `ws.OnBinaryMessage` called `IOnMessageFastPath(m)` with no session at all. * that mattered because retirement is not instant. `RetireOldSessionAsync` runs fire-and-forget beside the new connection and closes the old socket *gracefully*, so it keeps delivering for the length of the close handshake with its callback still attached. Whatever it sent landed in the new session's queue * after a reconnect those frames are stale. After a `ChangeServer` between networks they are worse than stale: switching mainnet to testnet, a handler that believes it is on testnet could receive the tail of mainnet's stream - transactions for accounts that do not exist there, ledger indexes from another chain, nothing marking them as belonging to the previous connection - * the session now travels with the frame and is checked before the write. A `null` session means the caller has none to name (`OnMessage`, which anyone may call), and nothing is rejected in that case + * the session now travels with the frame and is checked before the write. Matching the id alone would not do: both paths call `MarkAsRetiring()` while the session is still the active one and only `ConnectInternalAsync` installs its replacement, so frames arriving in that window carry an id that matches. The retiring flag is part of the test, under `_sessionLock` - the same guard `OnceOpen` and the other lifecycle callbacks already use, and the only way `IsRetiring` is published at all + * a `null` session means the caller has none to name (`OnMessage`, which anyone may call), and nothing is rejected in that case * `Connection.StaleSessionFramesDropped` counts what was discarded, kept separate from `DroppedStreamMessages` because the two mean different things: a non-zero value here is normal right after a reconnect, while the other means consumers are falling behind ## 10.12.0.0 08/16/2026 diff --git a/Tests/Xrpl.Tests/Client/TestUStaleSessionFrames.cs b/Tests/Xrpl.Tests/Client/TestUStaleSessionFrames.cs index bca7dd0e..a1382ba8 100644 --- a/Tests/Xrpl.Tests/Client/TestUStaleSessionFrames.cs +++ b/Tests/Xrpl.Tests/Client/TestUStaleSessionFrames.cs @@ -77,6 +77,49 @@ await client.connection.IOnMessageFastPath( } } + /// + /// A frame carrying the id of the session that is being retired is dropped too. + /// + /// + /// This is the case a plain id comparison misses. ChangeServer and the reconnect loop + /// mark the session retiring while it is still the active one - its replacement is installed + /// later, by ConnectInternalAsync - so frames arriving in that window carry an id that + /// matches. Whether the new connection is up yet decides nothing about whether these frames + /// belong to it. + /// + [TestMethod] + public async Task TestUFrameFromTheSessionBeingRetiredIsDropped() + { + using ScriptedResponseServer server = new ScriptedResponseServer(ScriptedReply); + using XrplClient client = new XrplClient(server.Url); + await client.Connect(); + + int calls = 0; + client.OnTransaction += _ => + { + calls++; + return Task.CompletedTask; + }; + + try + { + long? retiringSessionId = client.connection.MarkActiveSessionRetiringForTests(); + Assert.IsNotNull(retiringSessionId, "a connected client must have a session to retire"); + + await client.connection.IOnMessageFastPath( + Encoding.UTF8.GetBytes(TransactionMessage), + retiringSessionId); + + Assert.AreEqual(1L, client.connection.StaleSessionFramesDropped, + "an id that matches a retiring session is not an id that matches the live one"); + Assert.AreEqual(0, calls, "the handler saw a frame from the session being retired"); + } + finally + { + await client.Disconnect(); + } + } + /// /// The guard must not reject frames from the live session, or the stream stops entirely. /// diff --git a/Xrpl/Client/connection.cs b/Xrpl/Client/connection.cs index 487a0880..5fbd6b76 100644 --- a/Xrpl/Client/connection.cs +++ b/Xrpl/Client/connection.cs @@ -438,6 +438,26 @@ private static WebSocketClient CreateWebSocket(string url, ConnectionOptions con /// public long StaleSessionFramesDropped => Interlocked.Read(ref _staleSessionFramesDropped); + /// + /// Marks the session serving this connection as retiring and returns its id, reproducing the + /// window ChangeServer and the reconnect loop open between MarkAsRetiring() and + /// the installation of the replacement session. + /// + /// + /// Exists for tests: that window is reachable in production only by racing a real reconnect, + /// and the id of the session being retired is not otherwise observable. Marks it exactly the + /// way the two production paths do, under _sessionLock. + /// + /// The id of the session that was marked, or if there is none. + internal long? MarkActiveSessionRetiringForTests() + { + lock (_sessionLock) + { + _activeSession?.MarkAsRetiring(); + return _activeSession?.SessionId; + } + } + /// /// How many stream messages have been discarded because the consumer fell behind. /// @@ -3462,14 +3482,32 @@ private void EnqueueStreamMessage(byte[] frame, long? sessionId = null) // close runs fire-and-forget alongside the new connection. Without this check its last // frames land in the new session's queue and reach handlers as if they were current - // stale after a reconnect, and from an entirely different chain after a ChangeServer - // between networks. Lifecycle callbacks already guard the same way against _activeSession. + // between networks. + // + // Matching the id is not enough: ChangeServer and the reconnect loop call + // MarkAsRetiring() on the session while it is still _activeSession, and only + // ConnectInternalAsync installs its replacement. Frames arriving in that window carry the + // id of the very session being retired, so the retiring flag has to be part of the test - + // exactly as OnceOpen and the other lifecycle guards do it, and under the same lock, since + // IsRetiring is a plain bool published only by _sessionLock. // // A null sessionId means the caller cannot name a session (OnMessage, which anyone may // call): nothing to compare, so nothing is rejected. - if (sessionId is not null && _activeSession?.SessionId != sessionId) + if (sessionId is not null) { - Interlocked.Increment(ref _staleSessionFramesDropped); - return; + bool fromActiveSession; + lock (_sessionLock) + { + fromActiveSession = _activeSession != null && + _activeSession.SessionId == sessionId && + !_activeSession.IsRetiring; + } + + if (!fromActiveSession) + { + Interlocked.Increment(ref _staleSessionFramesDropped); + return; + } } { From 8bb44fc513c84b6eaadad392fdaa154d34773bc0 Mon Sep 17 00:00:00 2001 From: Aleksandr Platonenkov Date: Thu, 20 Aug 2026 11:55:23 -0300 Subject: [PATCH 05/16] =?UTF-8?q?fix(client):=20=D1=81=D0=B5=D1=81=D1=81?= =?UTF-8?q?=D0=B8=D1=8F=20=D0=B5=D0=B4=D0=B5=D1=82=20=D0=B2=20=D0=BE=D1=87?= =?UTF-8?q?=D0=B5=D1=80=D0=B5=D0=B4=D0=B8,=20=D0=B8=20=D1=81=D0=B2=D0=B5?= =?UTF-8?q?=D1=80=D0=BA=D0=B0=20=D0=B8=D0=B4=D1=91=D1=82=20=D0=BF=D0=B5?= =?UTF-8?q?=D1=80=D0=B5=D0=B4=20=D0=B2=D1=8B=D0=B7=D0=BE=D0=B2=D0=BE=D0=BC?= =?UTF-8?q?=20=D0=BE=D0=B1=D1=80=D0=B0=D0=B1=D0=BE=D1=82=D1=87=D0=B8=D0=BA?= =?UTF-8?q?=D0=BE=D0=B2?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Сверка только на входе в очередь ничего не гарантировала. Канал пересоздаётся под каждую сессию в StartMessageProcessor, и подмена идёт под _messageProcessorLock, а не под _sessionLock: между ответом «кадр от живой сессии» и записью в _streamMessageChannel сессия успевает уйти, а её замена — поставить новый канал, и кадр попадал уже в него. Окно шире, чем просто гонка двух строк: кадр, законно принятый у живой сессии, может пролежать в очереди до StreamMessageQueueCapacity кадров (10 000 по умолчанию) и быть выдан обработчикам после того, как ChangeServer увёл клиента в другую сеть. Поэтому сессия теперь едет вместе с кадром через очередь — Channel вместо Channel — и сверка идёт дважды: на входе, чтобы не занимать слот очереди, и в ProcessSessionFrameAsync прямо перед вызовом обработчиков. Второе и есть гарантия; первое — только экономия. Обе ветки, очередь и запасной путь fire-and-forget, идут через одну точку. Сама сверка вынесена в IsFromLiveSession — один замок на кадр против разбора JSON и вызова обработчика. Шов для тестов разделён на чтение и действие: ActiveSessionId и MarkActiveSessionRetiringForTests(). Новый тест детерминирован без гонок: процессор — единственный читатель и ждёт каждый обработчик, поэтому обработчик, который не возвращается, держит второй кадр в очереди сколько нужно. Проверено мутацией: снятие сверки на выдаче красит ровно его, снятие сверки на входе — два других. --- CHANGES.md | 3 +- .../Client/TestUStaleSessionFrames.cs | 83 ++++++++- Xrpl/Client/connection.cs | 165 +++++++++++++----- 3 files changed, 202 insertions(+), 49 deletions(-) diff --git a/CHANGES.md b/CHANGES.md index 8f80e3d9..0b84fabd 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -209,7 +209,8 @@ The entries below are grouped by what changed, not by the order the levels were * **Frames from a socket being retired no longer reach handlers as if they were current** (#112). Session identity was threaded into every lifecycle callback - `OnceOpen`, `OnConnectionFailed`, `OnceClose` all compare against `_activeSession.SessionId` - and the message path was the one exception: `ws.OnBinaryMessage` called `IOnMessageFastPath(m)` with no session at all. * that mattered because retirement is not instant. `RetireOldSessionAsync` runs fire-and-forget beside the new connection and closes the old socket *gracefully*, so it keeps delivering for the length of the close handshake with its callback still attached. Whatever it sent landed in the new session's queue * after a reconnect those frames are stale. After a `ChangeServer` between networks they are worse than stale: switching mainnet to testnet, a handler that believes it is on testnet could receive the tail of mainnet's stream - transactions for accounts that do not exist there, ledger indexes from another chain, nothing marking them as belonging to the previous connection - * the session now travels with the frame and is checked before the write. Matching the id alone would not do: both paths call `MarkAsRetiring()` while the session is still the active one and only `ConnectInternalAsync` installs its replacement, so frames arriving in that window carry an id that matches. The retiring flag is part of the test, under `_sessionLock` - the same guard `OnceOpen` and the other lifecycle callbacks already use, and the only way `IsRetiring` is published at all + * the session now travels with the frame - through the queue, not merely as far as it - and is checked twice: on the way in, to save a queue slot, and again immediately before handlers run, which is the check that guarantees anything. Checking only on the way in cannot: the channel is rebuilt per session under a different lock, and a frame accepted for the live session can still be dequeued long after that session is gone - the queue holds up to `StreamMessageQueueCapacity` frames (10 000 by default) + * matching the id alone would not do either: both paths call `MarkAsRetiring()` while the session is still the active one and only `ConnectInternalAsync` installs its replacement, so frames arriving in that window carry an id that matches. The retiring flag is part of the test, under `_sessionLock` - the same guard `OnceOpen` and the other lifecycle callbacks already use, and the only way `IsRetiring` is published at all * a `null` session means the caller has none to name (`OnMessage`, which anyone may call), and nothing is rejected in that case * `Connection.StaleSessionFramesDropped` counts what was discarded, kept separate from `DroppedStreamMessages` because the two mean different things: a non-zero value here is normal right after a reconnect, while the other means consumers are falling behind diff --git a/Tests/Xrpl.Tests/Client/TestUStaleSessionFrames.cs b/Tests/Xrpl.Tests/Client/TestUStaleSessionFrames.cs index a1382ba8..d29bd96a 100644 --- a/Tests/Xrpl.Tests/Client/TestUStaleSessionFrames.cs +++ b/Tests/Xrpl.Tests/Client/TestUStaleSessionFrames.cs @@ -2,6 +2,7 @@ using System; using System.Text; +using System.Threading; using System.Threading.Tasks; using Xrpl.Client; @@ -103,8 +104,9 @@ public async Task TestUFrameFromTheSessionBeingRetiredIsDropped() try { - long? retiringSessionId = client.connection.MarkActiveSessionRetiringForTests(); + long? retiringSessionId = client.connection.ActiveSessionId; Assert.IsNotNull(retiringSessionId, "a connected client must have a session to retire"); + client.connection.MarkActiveSessionRetiringForTests(); await client.connection.IOnMessageFastPath( Encoding.UTF8.GetBytes(TransactionMessage), @@ -120,6 +122,85 @@ await client.connection.IOnMessageFastPath( } } + /// + /// A frame that was queued while its session was live is still dropped if the session stops + /// being live before the frame is dispatched. + /// + /// + /// Checking on the way into the queue cannot be the guarantee. The queue holds up to + /// StreamMessageQueueCapacity frames (10 000 by default) and the channel itself is + /// rebuilt per session under a different lock, so between the check and the handler call the + /// session can retire and the client can be on another network entirely. + /// + /// Made deterministic by stalling the reader: the processor is a single reader that awaits + /// each handler, so a handler that does not return holds the second frame in the queue for as + /// long as the test needs. + /// + /// + [TestMethod] + public async Task TestUFrameQueuedBeforeRetirementIsNotDispatchedAfterIt() + { + using ScriptedResponseServer server = new ScriptedResponseServer(ScriptedReply); + using XrplClient client = new XrplClient(server.Url); + await client.Connect(); + + TaskCompletionSource firstFrameEntered = new TaskCompletionSource( + TaskCreationOptions.RunContinuationsAsynchronously); + TaskCompletionSource releaseFirstFrame = new TaskCompletionSource( + TaskCreationOptions.RunContinuationsAsynchronously); + int calls = 0; + + client.OnTransaction += async _ => + { + if (Interlocked.Increment(ref calls) == 1) + { + firstFrameEntered.TrySetResult(); + await releaseFirstFrame.Task; + } + }; + + try + { + long? sessionId = client.connection.ActiveSessionId; + Assert.IsNotNull(sessionId, "a connected client must have a session"); + + // Frame one occupies the reader and parks it inside the handler. + await client.connection.IOnMessageFastPath( + Encoding.UTF8.GetBytes(TransactionMessage), sessionId); + + Task entered = await Task.WhenAny(firstFrameEntered.Task, Task.Delay(TimeSpan.FromSeconds(5))); + Assert.AreSame(firstFrameEntered.Task, entered, "the reader never reached the handler"); + + // Frame two is accepted into the queue - the session is still live at this point - and + // waits there because the reader is parked. + await client.connection.IOnMessageFastPath( + Encoding.UTF8.GetBytes(TransactionMessage), sessionId); + Assert.AreEqual(0L, client.connection.StaleSessionFramesDropped, + "both frames were queued while the session was live"); + + // Only now does the session stop being the live one. + client.connection.MarkActiveSessionRetiringForTests(); + releaseFirstFrame.TrySetResult(); + + // The reader wakes, takes frame two and must refuse it. + DateTime deadline = DateTime.UtcNow.AddSeconds(5); + while (client.connection.StaleSessionFramesDropped == 0 && DateTime.UtcNow < deadline) + { + await Task.Delay(20); + } + + Assert.AreEqual(1L, client.connection.StaleSessionFramesDropped, + "a frame dequeued after its session retired must not be dispatched"); + Assert.AreEqual(1, Volatile.Read(ref calls), + "the handler saw a frame belonging to a session the client had already left"); + } + finally + { + releaseFirstFrame.TrySetResult(); + await client.Disconnect(); + } + } + /// /// The guard must not reject frames from the live session, or the stream stops entirely. /// diff --git a/Xrpl/Client/connection.cs b/Xrpl/Client/connection.cs index 5fbd6b76..4a60368c 100644 --- a/Xrpl/Client/connection.cs +++ b/Xrpl/Client/connection.cs @@ -418,7 +418,23 @@ private static WebSocketClient CreateWebSocket(string url, ConnectionOptions con // Carries the raw frame (byte[]), not text: stream events pair themselves with it the same // way a query response does, through AttachFrame, so their Raw/RawTransaction is available // without a second UTF-8 encode of a string that was itself decoded from these same bytes. - private Channel? _streamMessageChannel = null; + private Channel? _streamMessageChannel = null; + + /// + /// A stream frame together with the session whose socket produced it. + /// + /// + /// The session has to ride along in the queue, not just be checked on the way in: the channel + /// is rebuilt per session by , and between the check and + /// the write nothing holds the two together. Carrying the id lets the reader ask the question + /// again at the only moment that decides anything - just before handlers run. + /// + /// The raw UTF-8 frame. + /// + /// The session whose socket produced it, or when the caller had none to + /// name. + /// + private readonly record struct SessionFrame(byte[] Frame, long? SessionId); private long _droppedStreamMessages; @@ -439,22 +455,37 @@ private static WebSocketClient CreateWebSocket(string url, ConnectionOptions con public long StaleSessionFramesDropped => Interlocked.Read(ref _staleSessionFramesDropped); /// - /// Marks the session serving this connection as retiring and returns its id, reproducing the - /// window ChangeServer and the reconnect loop open between MarkAsRetiring() and - /// the installation of the replacement session. + /// The id of the session serving this connection, or if there is none. + /// + /// + /// Exists for tests, which need to name the session a frame came from - something only the + /// socket callbacks can otherwise do. + /// + internal long? ActiveSessionId + { + get + { + lock (_sessionLock) + { + return _activeSession?.SessionId; + } + } + } + + /// + /// Marks the session serving this connection as retiring, reproducing the window + /// ChangeServer and the reconnect loop open between MarkAsRetiring() and the + /// installation of the replacement session. /// /// - /// Exists for tests: that window is reachable in production only by racing a real reconnect, - /// and the id of the session being retired is not otherwise observable. Marks it exactly the - /// way the two production paths do, under _sessionLock. + /// Exists for tests: in production that window is reachable only by racing a real reconnect. + /// Marks it exactly the way the two production paths do, under _sessionLock. /// - /// The id of the session that was marked, or if there is none. - internal long? MarkActiveSessionRetiringForTests() + internal void MarkActiveSessionRetiringForTests() { lock (_sessionLock) { _activeSession?.MarkAsRetiring(); - return _activeSession?.SessionId; } } @@ -2996,7 +3027,7 @@ private void StartMessageProcessor() // itemDropped runs inside TryWrite, i.e. on the receive loop, so it does no more than // increment: raising an event or logging here would put consumer code back on the path // this channel exists to keep it off. Callers read DroppedStreamMessages instead. - _streamMessageChannel = System.Threading.Channels.Channel.CreateBounded( + _streamMessageChannel = System.Threading.Channels.Channel.CreateBounded( new BoundedChannelOptions(Math.Max(1, config?.StreamMessageQueueCapacity ?? 10000)) { SingleReader = true, @@ -3017,18 +3048,18 @@ private void StartMessageProcessor() var reader = channel.Reader; while (await reader.WaitToReadAsync(cts.Token).ConfigureAwait(false)) { - while (reader.TryRead(out var frame)) + while (reader.TryRead(out SessionFrame item)) { if (cts.Token.IsCancellationRequested) return; try { - await ProcessStreamMessageAsync(frame).ConfigureAwait(false); + await ProcessSessionFrameAsync(item).ConfigureAwait(false); } catch (Exception ex) { - await NotifyStreamProcessingErrorAsync(ex, frame).ConfigureAwait(false); + await NotifyStreamProcessingErrorAsync(ex, item.Frame).ConfigureAwait(false); } } } @@ -3476,40 +3507,57 @@ string Text() /// torn down again moments later. Untangling that is tracked separately. /// /// + /// + /// Whether a frame carrying this session id belongs to the connection as it stands right now. + /// + /// + /// Matching the id is not enough: ChangeServer and the reconnect loop call + /// MarkAsRetiring() on the session while it is still _activeSession, and only + /// ConnectInternalAsync installs its replacement. Frames arriving in that window carry + /// the id of the very session being retired, so the retiring flag is part of the test - as + /// OnceOpen and the other lifecycle guards do it, and under the same lock, since + /// IsRetiring is a plain bool published only by _sessionLock. + /// + /// A id means the caller cannot name a session + /// (, which anyone may call): nothing to compare, so nothing is + /// rejected. + /// + /// + private bool IsFromLiveSession(long? sessionId) + { + if (sessionId is null) + { + return true; + } + + lock (_sessionLock) + { + return _activeSession != null && + _activeSession.SessionId == sessionId && + !_activeSession.IsRetiring; + } + } + private void EnqueueStreamMessage(byte[] frame, long? sessionId = null) { // A retiring socket keeps delivering while InitiateGracefulCloseAsync completes, and that - // close runs fire-and-forget alongside the new connection. Without this check its last - // frames land in the new session's queue and reach handlers as if they were current - - // stale after a reconnect, and from an entirely different chain after a ChangeServer - // between networks. - // - // Matching the id is not enough: ChangeServer and the reconnect loop call - // MarkAsRetiring() on the session while it is still _activeSession, and only - // ConnectInternalAsync installs its replacement. Frames arriving in that window carry the - // id of the very session being retired, so the retiring flag has to be part of the test - - // exactly as OnceOpen and the other lifecycle guards do it, and under the same lock, since - // IsRetiring is a plain bool published only by _sessionLock. + // close runs fire-and-forget alongside the new connection. Without this its last frames + // reach handlers as if they were current - stale after a reconnect, and from an entirely + // different chain after a ChangeServer between networks. // - // A null sessionId means the caller cannot name a session (OnMessage, which anyone may - // call): nothing to compare, so nothing is rejected. - if (sessionId is not null) - { - bool fromActiveSession; - lock (_sessionLock) - { - fromActiveSession = _activeSession != null && - _activeSession.SessionId == sessionId && - !_activeSession.IsRetiring; - } - - if (!fromActiveSession) - { - Interlocked.Increment(ref _staleSessionFramesDropped); - return; - } + // Checking here only saves a queue slot. It cannot be the guarantee: the channel is + // rebuilt per session and the swap is under _messageProcessorLock, not _sessionLock, so + // between this answer and the write below the session can retire and its replacement + // install a new channel - and the frame would land in that one. The answer that counts is + // the one the reader asks in ProcessSessionFrameAsync, immediately before handlers run. + if (!IsFromLiveSession(sessionId)) + { + Interlocked.Increment(ref _staleSessionFramesDropped); + return; } + SessionFrame item = new SessionFrame(frame, sessionId); + { var channel = _streamMessageChannel; if (channel != null) @@ -3517,28 +3565,51 @@ private void EnqueueStreamMessage(byte[] frame, long? sessionId = null) // No failure branch on purpose: the channel is bounded with DropOldest, so // TryWrite always succeeds and silently evicts the oldest frame instead - counted // by the itemDropped callback rather than reported here. - channel.Writer.TryWrite(frame); + channel.Writer.TryWrite(item); } else { - _ = ProcessStreamMessageFireAndForgetAsync(frame); + _ = ProcessStreamMessageFireAndForgetAsync(item); } } } + /// + /// Runs a queued frame through the stream handlers, unless its session stopped being the live + /// one while it waited. + /// + /// + /// This is where the session check has to be final. A frame can be queued while its session is + /// live and dequeued after ChangeServer has moved the client to another network + /// entirely - and the queue holds up to + /// frames, so the wait is not + /// necessarily short. Asking again here costs one uncontended lock per frame, against a JSON + /// parse and a handler call. + /// + private Task ProcessSessionFrameAsync(SessionFrame item) + { + if (!IsFromLiveSession(item.SessionId)) + { + Interlocked.Increment(ref _staleSessionFramesDropped); + return Task.CompletedTask; + } + + return ProcessStreamMessageAsync(item.Frame); + } + /// /// Fire-and-forget stream message processing for single-threaded environments like WebAssembly. /// Uses ConfigureAwait(false) to prevent deadlocks and allow proper continuation scheduling. /// - private async Task ProcessStreamMessageFireAndForgetAsync(byte[] frame) + private async Task ProcessStreamMessageFireAndForgetAsync(SessionFrame item) { try { - await ProcessStreamMessageAsync(frame).ConfigureAwait(false); + await ProcessSessionFrameAsync(item).ConfigureAwait(false); } catch (Exception ex) { - await NotifyStreamProcessingErrorAsync(ex, frame).ConfigureAwait(false); + await NotifyStreamProcessingErrorAsync(ex, item.Frame).ConfigureAwait(false); } } From 4ecc2806d03123ebaaf0cf675fbe6b4545f345c9 Mon Sep 17 00:00:00 2001 From: Aleksandr Platonenkov Date: Thu, 20 Aug 2026 12:11:54 -0300 Subject: [PATCH 06/16] =?UTF-8?q?fix(client):=20=D0=BE=D1=82=D0=BA=D0=B0?= =?UTF-8?q?=D0=B7=20=D0=BA=D0=B0=D0=BD=D0=B0=D0=BB=D0=B0=20=D0=BD=D0=B5=20?= =?UTF-8?q?=D1=82=D0=B5=D1=80=D1=8F=D0=B5=D1=82=20=D0=BA=D0=B0=D0=B4=D1=80?= =?UTF-8?q?,=20=D1=82=D0=B5=D1=81=D1=82=20=D0=BD=D0=B5=20=D0=B7=D0=B0?= =?UTF-8?q?=D0=B2=D0=B8=D1=81=D0=B8=D1=82=20=D0=BE=D1=82=20=D1=81=D1=82?= =?UTF-8?q?=D0=B0=D1=80=D1=82=D0=B0=20=D0=BF=D1=80=D0=BE=D1=86=D0=B5=D1=81?= =?UTF-8?q?=D1=81=D0=BE=D1=80=D0=B0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit TryWrite звался ради побочного эффекта, а результат игнорировался — на том основании, что канал с DropOldest не отказывает. Для полной очереди это верно: она вытесняет старейший кадр, считает его через itemDropped и возвращает успех. Для завершённой — неверно. StopMessageProcessorInternal завершает writer уже после того, как обнулил _streamMessageChannel, поэтому тот, кто взял ссылку мгновением раньше, пишет в закрытый канал: TryWrite возвращает false, и кадр исчезал бесследно. Это не угол: StartPingTimer гасит процессор, а StartMessageProcessor поднимает его заново на каждом подключении, так что окно повторяется. Теперь отказ уводит кадр на запасной путь, где он всё так же проходит сверку сессии — очередь и запасной путь сходятся в ProcessSessionFrameAsync. Отдельно: тест на очередь опирался на то, что после Connect процессор уже поднят, а это не гарантировано вовсе. OnceOpen сначала отпускает ожидающих через connectionManager.ResolveAllAwaiting(), и только потом зовёт колбэк OnConnected и запускает процессор — то самое стартовое окно, описанное на EnqueueStreamMessage. В 20 локальных прогонах оно не проявилось, но порядок в коде именно такой, и тест теперь ждёт готовности явно, а не полагается на удачу. Швы: IsMessageProcessorRunning и CompleteStreamChannelWriterForTests(). Проверено мутацией: возврат к игнорированию отказа TryWrite красит ровно новый тест. --- CHANGES.md | 1 + .../Client/TestUStaleSessionFrames.cs | 72 +++++++++++++++++++ Xrpl/Client/connection.cs | 50 ++++++++++--- 3 files changed, 114 insertions(+), 9 deletions(-) diff --git a/CHANGES.md b/CHANGES.md index 0b84fabd..7796b6f4 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -212,6 +212,7 @@ The entries below are grouped by what changed, not by the order the levels were * the session now travels with the frame - through the queue, not merely as far as it - and is checked twice: on the way in, to save a queue slot, and again immediately before handlers run, which is the check that guarantees anything. Checking only on the way in cannot: the channel is rebuilt per session under a different lock, and a frame accepted for the live session can still be dequeued long after that session is gone - the queue holds up to `StreamMessageQueueCapacity` frames (10 000 by default) * matching the id alone would not do either: both paths call `MarkAsRetiring()` while the session is still the active one and only `ConnectInternalAsync` installs its replacement, so frames arriving in that window carry an id that matches. The retiring flag is part of the test, under `_sessionLock` - the same guard `OnceOpen` and the other lifecycle callbacks already use, and the only way `IsRetiring` is published at all * a `null` session means the caller has none to name (`OnMessage`, which anyone may call), and nothing is rejected in that case + * **a frame the channel refuses no longer vanishes.** `TryWrite` was called for its side effect and its result ignored, on the reasoning that a `DropOldest` channel never refuses - which is true of a full queue (it evicts, counts through `itemDropped` and reports success) and false of a completed one. `StopMessageProcessorInternal` completes the writer *after* clearing `_streamMessageChannel`, so whoever read the reference an instant earlier writes into a closed channel and the frame was dropped with nothing to show for it. Not a corner case: `StartPingTimer` tears the processor down and `StartMessageProcessor` builds it again on every connect. Such a frame now takes the fallback path, where it still faces the session check * `Connection.StaleSessionFramesDropped` counts what was discarded, kept separate from `DroppedStreamMessages` because the two mean different things: a non-zero value here is normal right after a reconnect, while the other means consumers are falling behind ## 10.12.0.0 08/16/2026 diff --git a/Tests/Xrpl.Tests/Client/TestUStaleSessionFrames.cs b/Tests/Xrpl.Tests/Client/TestUStaleSessionFrames.cs index d29bd96a..b756d4e8 100644 --- a/Tests/Xrpl.Tests/Client/TestUStaleSessionFrames.cs +++ b/Tests/Xrpl.Tests/Client/TestUStaleSessionFrames.cs @@ -43,6 +43,26 @@ public class TestUStaleSessionFrames } """; + /// + /// 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. + /// + private static async Task WaitForMessageProcessor(XrplClient client) + { + DateTime deadline = DateTime.UtcNow.AddSeconds(5); + while (!client.connection.IsMessageProcessorRunning && DateTime.UtcNow < deadline) + { + await Task.Delay(10); + } + + Assert.IsTrue(client.connection.IsMessageProcessorRunning, + "the message processor never started"); + } + /// /// A frame tagged with a session that is not the active one is dropped and counted. /// @@ -143,6 +163,7 @@ public async Task TestUFrameQueuedBeforeRetirementIsNotDispatchedAfterIt() using ScriptedResponseServer server = new ScriptedResponseServer(ScriptedReply); using XrplClient client = new XrplClient(server.Url); await client.Connect(); + await WaitForMessageProcessor(client); TaskCompletionSource firstFrameEntered = new TaskCompletionSource( TaskCreationOptions.RunContinuationsAsynchronously); @@ -201,6 +222,57 @@ await client.connection.IOnMessageFastPath( } } + /// + /// A frame the channel refuses is still delivered, through the fallback path. + /// + /// + /// The bounded channel uses DropOldest, so a full queue is not a refusal - it evicts and + /// reports success. TryWrite returns only for a completed + /// writer, which StopMessageProcessorInternal produces after clearing the field: anyone + /// holding the reference from an instant earlier writes into a closed channel. Since + /// StartPingTimer stops the processor and StartMessageProcessor rebuilds it on + /// every connect, this is the ordinary path rather than a corner of it, and a frame lost here + /// would be lost silently. + /// + [TestMethod] + public async Task TestUFrameRefusedByACompletedChannelTakesTheFallbackPath() + { + using ScriptedResponseServer server = new ScriptedResponseServer(ScriptedReply); + using XrplClient client = new XrplClient(server.Url); + await client.Connect(); + await WaitForMessageProcessor(client); + + TaskCompletionSource received = new TaskCompletionSource( + TaskCreationOptions.RunContinuationsAsynchronously); + client.OnTransaction += r => + { + received.TrySetResult(r); + return Task.CompletedTask; + }; + + try + { + long? sessionId = client.connection.ActiveSessionId; + Assert.IsNotNull(sessionId, "a connected client must have a session"); + + client.connection.CompleteStreamChannelWriterForTests(); + + await client.connection.IOnMessageFastPath( + Encoding.UTF8.GetBytes(TransactionMessage), sessionId); + + Task completed = await Task.WhenAny(received.Task, Task.Delay(TimeSpan.FromSeconds(5))); + Assert.AreSame(received.Task, completed, + "a frame the channel refused was neither queued nor dispatched - it vanished"); + + Assert.AreEqual(0L, client.connection.StaleSessionFramesDropped, + "the session was live throughout; nothing here is stale"); + } + finally + { + await client.Disconnect(); + } + } + /// /// The guard must not reject frames from the live session, or the stream stops entirely. /// diff --git a/Xrpl/Client/connection.cs b/Xrpl/Client/connection.cs index 4a60368c..d871029f 100644 --- a/Xrpl/Client/connection.cs +++ b/Xrpl/Client/connection.cs @@ -472,6 +472,33 @@ internal long? ActiveSessionId } } + /// + /// Whether the background message processor is up, i.e. whether stream frames are queued + /// 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 + /// . + /// + internal bool IsMessageProcessorRunning => _streamMessageChannel != null; + + /// + /// Completes the stream channel's writer without clearing the channel, reproducing the state + /// StopMessageProcessorInternal leaves behind for anyone who read + /// _streamMessageChannel just before it was cleared. + /// + /// + /// Exists for tests: in production that state lasts between one field read and one call, and + /// is not reachable deliberately. + /// + internal void CompleteStreamChannelWriterForTests() + { + _streamMessageChannel?.Writer.Complete(); + } + /// /// Marks the session serving this connection as retiring, reproducing the window /// ChangeServer and the reconnect loop open between MarkAsRetiring() and the @@ -3559,18 +3586,23 @@ private void EnqueueStreamMessage(byte[] frame, long? sessionId = null) SessionFrame item = new SessionFrame(frame, sessionId); { + // The channel is bounded with DropOldest, so a full queue is not a refusal: TryWrite + // evicts the oldest frame, counts it through itemDropped and reports success. It + // refuses only a completed writer - and that happens on the ordinary path, not just in + // some corner: StopMessageProcessorInternal completes the writer after clearing + // _streamMessageChannel, so a reader that got the reference an instant earlier writes + // into a channel that is already closed. StartPingTimer tears the processor down and + // StartMessageProcessor builds it again on every connect, so the window recurs. + // + // A refused frame therefore goes down the fallback rather than disappearing. It still + // faces the session check there - fallback and queue meet in ProcessSessionFrameAsync. var channel = _streamMessageChannel; - if (channel != null) + if (channel?.Writer.TryWrite(item) == true) { - // No failure branch on purpose: the channel is bounded with DropOldest, so - // TryWrite always succeeds and silently evicts the oldest frame instead - counted - // by the itemDropped callback rather than reported here. - channel.Writer.TryWrite(item); - } - else - { - _ = ProcessStreamMessageFireAndForgetAsync(item); + return; } + + _ = ProcessStreamMessageFireAndForgetAsync(item); } } From bffd86d14896133f67d61fe75721ab23af7aa773 Mon Sep 17 00:00:00 2001 From: Aleksandr Platonenkov Date: Thu, 20 Aug 2026 12:22:18 -0300 Subject: [PATCH 07/16] =?UTF-8?q?refactor(client):=20=D1=81=D0=B5=D0=BB?= =?UTF-8?q?=D1=84-=D1=80=D0=B5=D0=B2=D1=8C=D1=8E=20=E2=80=94=20=D1=80?= =?UTF-8?q?=D0=B0=D1=81=D0=BA=D0=BB=D0=B0=D0=B4=D0=BA=D0=B0=20=D0=B4=D0=BE?= =?UTF-8?q?=D0=BA=D1=83=D0=BC=D0=B5=D0=BD=D1=82=D0=B0=D1=86=D0=B8=D0=B8,?= =?UTF-8?q?=20=D0=B1=D0=B0=D1=80=D1=8C=D0=B5=D1=80,=20=D1=81=D1=87=D1=91?= =?UTF-8?q?=D1=82=D1=87=D0=B8=D0=BA=20=D0=B2=20=D0=B8=D0=BD=D1=82=D0=B5?= =?UTF-8?q?=D1=80=D1=84=D0=B5=D0=B9=D1=81=D0=B5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Проверка собственного диффа поверх origin/dev. Дефект: doc-комментарий EnqueueStreamMessage оказался оторван от метода — IsFromLiveSession вставился между и сигнатурой, так что у предиката стало два , а сам EnqueueStreamMessage остался вовсе без документации. Восстановлено. StaleSessionFramesDropped проброшен в IXrplClient рядом с DroppedStreamMessages: счётчик заводился ради наблюдаемости, а прочитать его через интерфейс было нельзя. Это новый член публичного интерфейса — внешним реализациям придётся его добавить; PR и так помечен как ломающий. IsMessageProcessorRunning читает поле через Volatile.Read: поле пишется под _messageProcessorLock, и после того как я требовал замок для IsRetiring, обычное чтение здесь было бы двойным стандартом. Полноценный замок брать нельзя — StopMessageProcessorInternal держит его до двух секунд на ожидании задачи читателя. Мелочи: комментарий над полем канала говорил «byte[]», хотя канал уже несёт SessionFrame; объявление SessionFrame больше не разрывает группу полей, а стоит рядом с предикатом, который его читает; убран остаточный блок из EnqueueStreamMessage; описание ProcessStreamMessageFireAndForgetAsync больше не называет путь браузерным — входов в него теперь три, и ни один не зависит от платформы. --- CHANGES.md | 2 +- Tests/Xrpl.Tests/Sugar/TestUAutofillFees.cs | 2 + Xrpl/Client/IXrplClient.cs | 17 +++ Xrpl/Client/connection.cs | 134 +++++++++++--------- 4 files changed, 92 insertions(+), 63 deletions(-) diff --git a/CHANGES.md b/CHANGES.md index 7796b6f4..9c6c6301 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -213,7 +213,7 @@ The entries below are grouped by what changed, not by the order the levels were * matching the id alone would not do either: both paths call `MarkAsRetiring()` while the session is still the active one and only `ConnectInternalAsync` installs its replacement, so frames arriving in that window carry an id that matches. The retiring flag is part of the test, under `_sessionLock` - the same guard `OnceOpen` and the other lifecycle callbacks already use, and the only way `IsRetiring` is published at all * a `null` session means the caller has none to name (`OnMessage`, which anyone may call), and nothing is rejected in that case * **a frame the channel refuses no longer vanishes.** `TryWrite` was called for its side effect and its result ignored, on the reasoning that a `DropOldest` channel never refuses - which is true of a full queue (it evicts, counts through `itemDropped` and reports success) and false of a completed one. `StopMessageProcessorInternal` completes the writer *after* clearing `_streamMessageChannel`, so whoever read the reference an instant earlier writes into a closed channel and the frame was dropped with nothing to show for it. Not a corner case: `StartPingTimer` tears the processor down and `StartMessageProcessor` builds it again on every connect. Such a frame now takes the fallback path, where it still faces the session check - * `Connection.StaleSessionFramesDropped` counts what was discarded, kept separate from `DroppedStreamMessages` because the two mean different things: a non-zero value here is normal right after a reconnect, while the other means consumers are falling behind + * `Connection.StaleSessionFramesDropped` counts what was discarded, kept separate from `DroppedStreamMessages` because the two mean different things: a non-zero value here is normal right after a reconnect, while the other means consumers are falling behind. Also on `IXrplClient`, next to `DroppedStreamMessages` - a counter nobody can read is not observability. **New interface member**: an external implementation of `IXrplClient` has to add it ## 10.12.0.0 08/16/2026 diff --git a/Tests/Xrpl.Tests/Sugar/TestUAutofillFees.cs b/Tests/Xrpl.Tests/Sugar/TestUAutofillFees.cs index c861f4cd..48a9c722 100644 --- a/Tests/Xrpl.Tests/Sugar/TestUAutofillFees.cs +++ b/Tests/Xrpl.Tests/Sugar/TestUAutofillFees.cs @@ -643,6 +643,8 @@ public FeeTestClient(string feeXrp, uint reserveInc, string maxFeeXRP = "5") public Connection connection => null!; public long DroppedStreamMessages => 0; + public long StaleSessionFramesDropped => 0; + // Declared, never raised: this client answers requests, it does not stream. Being able to // declare them at all is the point of the events moving onto IXrplClient - a substitute // client could not carry them while they lived only on Connection. diff --git a/Xrpl/Client/IXrplClient.cs b/Xrpl/Client/IXrplClient.cs index 19609135..649cc4a9 100644 --- a/Xrpl/Client/IXrplClient.cs +++ b/Xrpl/Client/IXrplClient.cs @@ -69,6 +69,20 @@ public interface IXrplClient : IDisposable /// long DroppedStreamMessages { get; } + /// + /// How many stream frames were discarded because they came from a connection this client + /// had already left. + /// + /// + /// A socket being retired keeps delivering until its graceful close finishes, so a few + /// frames can arrive after a reconnect or ChangeServer has moved on. Delivering them + /// would be wrong rather than merely late: after a change of network they describe a + /// different chain entirely. A non-zero value here is therefore normal right after a + /// reconnect and says nothing about handler speed - that is + /// , and the two are counted apart on purpose. + /// + long StaleSessionFramesDropped { get; } + /// Node error reported over the socket. event OnError OnError; @@ -559,6 +573,9 @@ public class ClientOptions : ConnectionOptions /// public long DroppedStreamMessages => connection.DroppedStreamMessages; + /// + public long StaleSessionFramesDropped => connection.StaleSessionFramesDropped; + // Forwarded, not relayed: add/remove reach the same Connection a caller would have used // through the property, so this type holds no delegates and no subscription of its own. A // relaying version - a local event plus a subscription to the connection that re-raises it diff --git a/Xrpl/Client/connection.cs b/Xrpl/Client/connection.cs index d871029f..9ea70d04 100644 --- a/Xrpl/Client/connection.cs +++ b/Xrpl/Client/connection.cs @@ -415,27 +415,12 @@ private static WebSocketClient CreateWebSocket(string url, ConnectionOptions con // to prevent head-of-line blocking that causes ping timeouts under high stream load // Channel is created per-session to prevent cross-session message leakage // Using Channel instead of BlockingCollection for true async support in WebAssembly - // Carries the raw frame (byte[]), not text: stream events pair themselves with it the same + // Carries the raw frame bytes, not text: stream events pair themselves with the frame the same // way a query response does, through AttachFrame, so their Raw/RawTransaction is available // without a second UTF-8 encode of a string that was itself decoded from these same bytes. + // The frame travels wrapped in SessionFrame, which names the session that produced it. private Channel? _streamMessageChannel = null; - /// - /// A stream frame together with the session whose socket produced it. - /// - /// - /// The session has to ride along in the queue, not just be checked on the way in: the channel - /// is rebuilt per session by , and between the check and - /// the write nothing holds the two together. Carrying the id lets the reader ask the question - /// again at the only moment that decides anything - just before handlers run. - /// - /// The raw UTF-8 frame. - /// - /// The session whose socket produced it, or when the caller had none to - /// name. - /// - private readonly record struct SessionFrame(byte[] Frame, long? SessionId); - private long _droppedStreamMessages; private long _staleSessionFramesDropped; @@ -483,7 +468,12 @@ internal long? ActiveSessionId /// be ahead of the queue. That ordering is the startup window described on /// . /// - internal bool IsMessageProcessorRunning => _streamMessageChannel != null; + /// + /// 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; /// /// Completes the stream channel's writer without clearing the channel, reproducing the state @@ -3506,34 +3496,21 @@ string Text() } /// - /// Hands a stream message to the background processor. + /// A stream frame together with the session whose socket produced it. /// /// - /// Used for ordinary stream messages (no id) and for follow-ups carrying an id - /// that matches no pending request, such as path_find updates. - /// - /// Browsers used to take a separate path here - one fire-and-forget task per frame, bypassing - /// the queue entirely, so did not - /// apply, stayed at zero however far handlers fell behind, - /// the backlog was bounded by nothing, and concurrent dispatch could hand handlers events out - /// of the order the node sent them. The queue was built for this environment in the first - /// place ("true async support in WebAssembly single-threaded environment" on - /// ), and measurement confirmed it works there: running the - /// Blazor demo against mainnet, the queue delivered 1 004 transactions over 52 s (19.2 tx/s, - /// 13 ledgers) with no console errors and timestamps in order - against 462 over 33 s - /// (13.9 tx/s) on the bypass. So the platforms no longer diverge: capacity, eviction counting - /// 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 session has to ride along in the queue, not just be checked on the way in: the channel + /// is rebuilt per session by , and between the check and + /// the write nothing holds the two together. Carrying the id lets the reader ask the question + /// again at the only moment that decides anything - just before handlers run. /// + /// The raw UTF-8 frame. + /// + /// The session whose socket produced it, or when the caller had none to + /// name. + /// + private readonly record struct SessionFrame(byte[] Frame, long? SessionId); + /// /// Whether a frame carrying this session id belongs to the connection as it stands right now. /// @@ -3565,6 +3542,35 @@ private bool IsFromLiveSession(long? sessionId) } } + /// + /// Hands a stream message to the background processor. + /// + /// + /// Used for ordinary stream messages (no id) and for follow-ups carrying an id + /// that matches no pending request, such as path_find updates. + /// + /// Browsers used to take a separate path here - one fire-and-forget task per frame, bypassing + /// the queue entirely, so did not + /// apply, stayed at zero however far handlers fell behind, + /// the backlog was bounded by nothing, and concurrent dispatch could hand handlers events out + /// of the order the node sent them. The queue was built for this environment in the first + /// place ("true async support in WebAssembly single-threaded environment" on + /// ), and measurement confirmed it works there: running the + /// Blazor demo against mainnet, the queue delivered 1 004 transactions over 52 s (19.2 tx/s, + /// 13 ledgers) with no console errors and timestamps in order - against 462 over 33 s + /// (13.9 tx/s) on the bypass. So the platforms no longer diverge: capacity, eviction counting + /// 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. + /// + /// private void EnqueueStreamMessage(byte[] frame, long? sessionId = null) { // A retiring socket keeps delivering while InitiateGracefulCloseAsync completes, and that @@ -3585,25 +3591,23 @@ private void EnqueueStreamMessage(byte[] frame, long? sessionId = null) SessionFrame item = new SessionFrame(frame, sessionId); + // The channel is bounded with DropOldest, so a full queue is not a refusal: TryWrite + // evicts the oldest frame, counts it through itemDropped and reports success. It + // refuses only a completed writer - and that happens on the ordinary path, not just in + // some corner: StopMessageProcessorInternal completes the writer after clearing + // _streamMessageChannel, so a reader that got the reference an instant earlier writes + // into a channel that is already closed. StartPingTimer tears the processor down and + // StartMessageProcessor builds it again on every connect, so the window recurs. + // + // A refused frame therefore goes down the fallback rather than disappearing. It still + // faces the session check there - fallback and queue meet in ProcessSessionFrameAsync. + Channel? channel = _streamMessageChannel; + if (channel?.Writer.TryWrite(item) == true) { - // The channel is bounded with DropOldest, so a full queue is not a refusal: TryWrite - // evicts the oldest frame, counts it through itemDropped and reports success. It - // refuses only a completed writer - and that happens on the ordinary path, not just in - // some corner: StopMessageProcessorInternal completes the writer after clearing - // _streamMessageChannel, so a reader that got the reference an instant earlier writes - // into a channel that is already closed. StartPingTimer tears the processor down and - // StartMessageProcessor builds it again on every connect, so the window recurs. - // - // A refused frame therefore goes down the fallback rather than disappearing. It still - // faces the session check there - fallback and queue meet in ProcessSessionFrameAsync. - var channel = _streamMessageChannel; - if (channel?.Writer.TryWrite(item) == true) - { - return; - } - - _ = ProcessStreamMessageFireAndForgetAsync(item); + return; } + + _ = ProcessStreamMessageFireAndForgetAsync(item); } /// @@ -3630,9 +3634,15 @@ private Task ProcessSessionFrameAsync(SessionFrame item) } /// - /// Fire-and-forget stream message processing for single-threaded environments like WebAssembly. - /// Uses ConfigureAwait(false) to prevent deadlocks and allow proper continuation scheduling. + /// Processes a stream frame outside the queue, on its own task. /// + /// + /// Three ways in, none of them platform-specific since browsers stopped taking a path of their + /// own: before has run, after the processor was stopped, + /// and when the channel refuses the write because its writer is already completed. Ordering + /// and the capacity bound do not hold here - that is the cost of not losing the frame. + /// ConfigureAwait(false) throughout, to keep continuations off a captured context. + /// private async Task ProcessStreamMessageFireAndForgetAsync(SessionFrame item) { try From 68087d4372fa460a02db3527ab1ea3153713c073 Mon Sep 17 00:00:00 2001 From: Aleksandr Platonenkov Date: Thu, 20 Aug 2026 12:45:16 -0300 Subject: [PATCH 08/16] =?UTF-8?q?fix(client):=20=D0=B7=D0=B0=D0=BF=D0=B0?= =?UTF-8?q?=D1=81=D0=BD=D0=BE=D0=B9=20=D0=BF=D1=83=D1=82=D1=8C=20=D0=BE?= =?UTF-8?q?=D1=82=D0=B4=D0=B0=D1=91=D1=82=20=D0=BA=D0=B0=D0=B4=D1=80=20?= =?UTF-8?q?=D0=B4=D0=BE=20=D1=80=D0=B0=D0=B1=D0=BE=D1=82=D1=8B,=20=D0=B0?= =?UTF-8?q?=20=D0=BD=D0=B5=20=D0=BF=D0=BE=D1=81=D0=BB=D0=B5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Асинхронный метод выполняется на потоке вызывающего до первого настоящего await, а первый настоящий await внутри ProcessStreamMessageAsync стоит уже после JsonSerializer.Deserialize. То есть каждый кадр, ушедший на запасной путь, разбирался на приёмном цикле — плюс всё, что обработчик успевал сделать до своего первого await. Это ровно та блокировка головы очереди, ради которой очередь и заведена, возвращённая для стартового окна, для остановленного процессора и для отказа канала. Yield в начале запасного пути её убирает. Следствие, которое стоит назвать вслух: OnMessage больше не выдаёт кадр синхронно, когда процессора нет. На пути через очередь он этого не делал никогда, так что теперь два пути ведут себя одинаково — но код, который звал OnMessage на неподключённом клиенте и читал состояние обработчика следующей строкой, опирался именно на разницу. На это опирались два теста форвардинга событий: они звали OnMessage и тут же проверяли счётчик. Переписаны на ожидание — со свидетелем, которого никто не снимает, иначе «счётчик всё ещё 1» после снятия обработчика проходило бы и для кадра, который просто ещё не обработан. Проверено мутацией: замена remove на += в аксессоре клиента по-прежнему красит зеркальный тест. Новый тест — TestUFallbackPathReturnsBeforeHandlersRun, детерминированный по построению: обработчик встаёт на блокирующее ожидание, и если бы кадр обрабатывался встроенно, вызов инъекции не вернулся бы до его освобождения. Проверено мутацией: снятие yield красит ровно его. --- CHANGES.md | 2 + .../Client/TestUClientStreamEvents.cs | 44 +++++++++++++ .../Client/TestUStaleSessionFrames.cs | 61 +++++++++++++++++++ Xrpl/Client/connection.cs | 12 ++++ 4 files changed, 119 insertions(+) diff --git a/CHANGES.md b/CHANGES.md index 9c6c6301..0ab16747 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -212,6 +212,8 @@ The entries below are grouped by what changed, not by the order the levels were * the session now travels with the frame - through the queue, not merely as far as it - and is checked twice: on the way in, to save a queue slot, and again immediately before handlers run, which is the check that guarantees anything. Checking only on the way in cannot: the channel is rebuilt per session under a different lock, and a frame accepted for the live session can still be dequeued long after that session is gone - the queue holds up to `StreamMessageQueueCapacity` frames (10 000 by default) * matching the id alone would not do either: both paths call `MarkAsRetiring()` while the session is still the active one and only `ConnectInternalAsync` installs its replacement, so frames arriving in that window carry an id that matches. The retiring flag is part of the test, under `_sessionLock` - the same guard `OnceOpen` and the other lifecycle callbacks already use, and the only way `IsRetiring` is published at all * a `null` session means the caller has none to name (`OnMessage`, which anyone may call), and nothing is rejected in that case + * consequence worth naming: `OnMessage` no longer dispatches inline when there is no processor. It never did on the queued path, so this makes the two agree - but code that called `OnMessage` on an unconnected client and read handler state on the next line was relying on the difference + * **the fallback path hands the frame off before doing any work.** An async method runs on its caller's thread up to the first real await, and the first real await inside `ProcessStreamMessageAsync` comes after `JsonSerializer.Deserialize` - so every frame taking the fallback had its JSON parsed on the receive loop, plus whatever a handler did before its own first await. That is the head-of-line blocking the queue exists to prevent, reintroduced for the startup window, for a stopped processor and for a refused write. A yield at the top of the fallback ends it * **a frame the channel refuses no longer vanishes.** `TryWrite` was called for its side effect and its result ignored, on the reasoning that a `DropOldest` channel never refuses - which is true of a full queue (it evicts, counts through `itemDropped` and reports success) and false of a completed one. `StopMessageProcessorInternal` completes the writer *after* clearing `_streamMessageChannel`, so whoever read the reference an instant earlier writes into a closed channel and the frame was dropped with nothing to show for it. Not a corner case: `StartPingTimer` tears the processor down and `StartMessageProcessor` builds it again on every connect. Such a frame now takes the fallback path, where it still faces the session check * `Connection.StaleSessionFramesDropped` counts what was discarded, kept separate from `DroppedStreamMessages` because the two mean different things: a non-zero value here is normal right after a reconnect, while the other means consumers are falling behind. Also on `IXrplClient`, next to `DroppedStreamMessages` - a counter nobody can read is not observability. **New interface member**: an external implementation of `IXrplClient` has to add it diff --git a/Tests/Xrpl.Tests/Client/TestUClientStreamEvents.cs b/Tests/Xrpl.Tests/Client/TestUClientStreamEvents.cs index 91f52501..a154e3d5 100644 --- a/Tests/Xrpl.Tests/Client/TestUClientStreamEvents.cs +++ b/Tests/Xrpl.Tests/Client/TestUClientStreamEvents.cs @@ -1,6 +1,7 @@ using Microsoft.VisualStudio.TestTools.UnitTesting; using System; +using System.Threading; using System.Threading.Tasks; using Xrpl.Client; @@ -36,6 +37,27 @@ public class TestUClientStreamEvents } """; + /// + /// Waits until the counter reaches , or fails. + /// + /// + /// OnMessage returning does not mean handlers have run: a stream frame is handed to the + /// background processor, and when there is none - an unconnected client, as here - to a task + /// that yields before doing anything, so that parsing and handler code never run on the + /// receive loop. Both paths are asynchronous, which is the point; the test has to wait rather + /// than assume. + /// + private static void WaitForCount(Func counter, int expected, string message) + { + DateTime deadline = DateTime.UtcNow.AddSeconds(5); + while (counter() < expected && DateTime.UtcNow < deadline) + { + Thread.Sleep(10); + } + + Assert.AreEqual(expected, counter(), message); + } + /// /// A handler registered through the interface receives the event, and the raw bytes come with /// it - the whole point of reaching the stream through the contract. @@ -90,12 +112,24 @@ public async Task TestUHandlerAddedThroughTheClientCanBeRemovedThroughTheConnect // that work: a relaying client would keep its own subscriber list, the removal would miss // it, and the handler would keep firing. Removing through the same surface it was added // to cannot tell the two apart - both pass - which is why the test crosses surfaces. + // A witness that is never removed, added after the handler so it runs after it: it tells + // the test when a frame has been dispatched. Without it, "calls is still 1" after the + // removal would pass just as well for a frame that has not been processed yet. + int dispatched = 0; + client.connection.OnTransaction += _ => + { + Interlocked.Increment(ref dispatched); + return Task.CompletedTask; + }; + contract.OnTransaction += handler; await client.connection.OnMessage(TransactionMessage); + WaitForCount(() => Volatile.Read(ref dispatched), 1, "the first frame was never dispatched"); Assert.AreEqual(1, calls, "sanity: the handler is attached"); client.connection.OnTransaction -= handler; await client.connection.OnMessage(TransactionMessage); + WaitForCount(() => Volatile.Read(ref dispatched), 2, "the second frame was never dispatched"); Assert.AreEqual(1, calls, "removing through the connection left the handler attached - the client is relaying into its own list rather than forwarding"); } @@ -123,11 +157,21 @@ public async Task TestUHandlerAddedThroughTheConnectionCanBeRemovedThroughTheCli }; client.connection.OnTransaction += handler; + + int dispatched = 0; + client.connection.OnTransaction += _ => + { + Interlocked.Increment(ref dispatched); + return Task.CompletedTask; + }; + await client.connection.OnMessage(TransactionMessage); + WaitForCount(() => Volatile.Read(ref dispatched), 1, "the first frame was never dispatched"); Assert.AreEqual(1, calls, "sanity: the handler is attached"); contract.OnTransaction -= handler; await client.connection.OnMessage(TransactionMessage); + WaitForCount(() => Volatile.Read(ref dispatched), 2, "the second frame was never dispatched"); Assert.AreEqual(1, calls, "removing through the client left the handler attached - its remove accessor does not reach the connection"); } diff --git a/Tests/Xrpl.Tests/Client/TestUStaleSessionFrames.cs b/Tests/Xrpl.Tests/Client/TestUStaleSessionFrames.cs index b756d4e8..e96de8e0 100644 --- a/Tests/Xrpl.Tests/Client/TestUStaleSessionFrames.cs +++ b/Tests/Xrpl.Tests/Client/TestUStaleSessionFrames.cs @@ -273,6 +273,67 @@ await client.connection.IOnMessageFastPath( } } + /// + /// The fallback path hands the frame off before doing any of the work, so the receive loop is + /// not the thread that parses JSON and runs handlers. + /// + /// + /// An async method runs on its caller's thread up to the first real await, and the first real + /// await inside ProcessStreamMessageAsync comes after JsonSerializer.Deserialize. + /// Without a yield at the top of the fallback, everything up to a handler's own first await + /// runs inline on the socket callback - the head-of-line blocking the queue exists to prevent, + /// reintroduced for the startup window, a stopped processor and a refused write. + /// + /// Deterministic by construction: the handler parks on a blocking wait. If the frame were + /// processed inline, the injecting call could not return until the handler was released, so + /// the wait below would time out rather than merely be slow. + /// + /// + [TestMethod] + public async Task TestUFallbackPathReturnsBeforeHandlersRun() + { + using ScriptedResponseServer server = new ScriptedResponseServer(ScriptedReply); + using XrplClient client = new XrplClient(server.Url); + await client.Connect(); + await WaitForMessageProcessor(client); + + using ManualResetEventSlim handlerEntered = new ManualResetEventSlim(initialState: false); + using ManualResetEventSlim releaseHandler = new ManualResetEventSlim(initialState: false); + + client.OnTransaction += _ => + { + handlerEntered.Set(); + releaseHandler.Wait(TimeSpan.FromSeconds(10)); + return Task.CompletedTask; + }; + + try + { + long? sessionId = client.connection.ActiveSessionId; + Assert.IsNotNull(sessionId, "a connected client must have a session"); + + // Completing the writer forces the next frame onto the fallback path. + client.connection.CompleteStreamChannelWriterForTests(); + + // Injected from a task of its own: were the frame processed inline, the call itself + // would block and there would be no Task to wait on. + Task inject = Task.Run(() => client.connection.IOnMessageFastPath( + Encoding.UTF8.GetBytes(TransactionMessage), sessionId)); + + Task finished = await Task.WhenAny(inject, Task.Delay(TimeSpan.FromSeconds(5))); + Assert.AreSame(inject, finished, + "the fallback ran the frame inline: injection did not return while the handler was parked"); + + Assert.IsTrue(handlerEntered.Wait(TimeSpan.FromSeconds(5)), + "the frame never reached the handler at all"); + } + finally + { + releaseHandler.Set(); + await client.Disconnect(); + } + } + /// /// The guard must not reject frames from the live session, or the stream stops entirely. /// diff --git a/Xrpl/Client/connection.cs b/Xrpl/Client/connection.cs index 9ea70d04..e1558973 100644 --- a/Xrpl/Client/connection.cs +++ b/Xrpl/Client/connection.cs @@ -3642,9 +3642,21 @@ private Task ProcessSessionFrameAsync(SessionFrame item) /// and when the channel refuses the write because its writer is already completed. Ordering /// and the capacity bound do not hold here - that is the cost of not losing the frame. /// ConfigureAwait(false) throughout, to keep continuations off a captured context. + /// + /// The yield is what makes "fire and forget" true. Without it an async method runs on the + /// caller's thread up to its first real await, and the first real await inside + /// comes after + /// JsonSerializer.Deserialize - so the receive loop would pay for parsing every frame + /// that takes this path, plus whatever a handler does before its own first await. That is + /// precisely the head-of-line blocking the queue exists to prevent, and the fallback would + /// have reintroduced it for the startup window, for a stopped processor and for a refused + /// write. + /// /// private async Task ProcessStreamMessageFireAndForgetAsync(SessionFrame item) { + await Task.Yield(); + try { await ProcessSessionFrameAsync(item).ConfigureAwait(false); From 6c8319bff7eaa093bae944dfeac3b1489d4f7642 Mon Sep 17 00:00:00 2001 From: Aleksandr Platonenkov Date: Thu, 20 Aug 2026 13:07:03 -0300 Subject: [PATCH 09/16] =?UTF-8?q?test(client):=20=D0=B4=D0=BE=D1=81=D1=82?= =?UTF-8?q?=D0=B0=D0=B2=D0=BA=D0=B0=20=D0=B8=D0=BC=D0=B5=D0=BD=D0=BE=D0=B2?= =?UTF-8?q?=D0=B0=D0=BD=D0=BD=D0=BE=D0=B9=20=D1=81=D0=B5=D1=81=D1=81=D0=B8?= =?UTF-8?q?=D0=B8=20=D0=BF=D1=80=D0=BE=D0=B2=D0=B5=D1=80=D1=8F=D0=B5=D1=82?= =?UTF-8?q?=D1=81=D1=8F=20=D0=BE=D1=82=D0=B4=D0=B5=D0=BB=D1=8C=D0=BD=D0=BE?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Тест на доставку звал OnMessage, то есть проверял только кадр без сессии. Реализация, отвергающая любую именованную сессию, его проходила — а это означало бы, что не доставляется вообще ничего: все кадры, которые производит сокет, именованы. Добавлен TestUFrameNamingTheLiveSessionIsDelivered, прежний переименован в TestUFrameNamingNoSessionIsDelivered — теперь имя говорит, какой именно случай он держит. Проверено мутацией: с return false для любой непустой сессии новый тест краснеет, а прежний остаётся зелёным — ровно тот разрыв, о котором шла речь. Заодно уточнена формулировка в CHANGES: утверждение «приёмный цикл делает только TryWrite» верно для кадра из очереди, но кадр на запасном пути отдаётся именно с приёмного цикла. Вывод не меняется — передача уступает поток до разбора, так что обработчики на приёмном цикле не выполняются, — но механизм описан теперь для обоих путей. --- CHANGES.md | 2 +- .../Client/TestUStaleSessionFrames.cs | 54 +++++++++++++++++-- 2 files changed, 51 insertions(+), 5 deletions(-) diff --git a/CHANGES.md b/CHANGES.md index 0ab16747..b17ef76e 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -205,7 +205,7 @@ The entries below are grouped by what changed, not by the order the levels were * `ConnectionOptions.StreamMessageQueueCapacity` (default 10 000, unchanged) sizes the queue - raise it for a consumer that must not miss events and can hold the frames, lower it to bound memory harder * the counter is incremented from `Channel`'s `itemDropped` callback, which runs inside `TryWrite` on the receive loop. It does nothing but increment for that reason: raising an event or logging there would put consumer code back on the path this queue exists to keep it off - the very failure #105 described but which the queue already prevented * **browsers no longer take a separate path** (#110). `EnqueueStreamMessage` used to start one fire-and-forget task per frame under WebAssembly, bypassing the queue: the capacity was not consulted, nothing was evicted, the counter stayed at zero however far handlers fell behind, the backlog was bounded by nothing, and concurrent dispatch could hand handlers events out of the order the node sent them. The queue was built for that environment to begin with - `StartMessageProcessor` says "true async support in WebAssembly single-threaded environment" - and measurement confirms it works there. Running the Blazor demo against mainnet: **1 004 transactions over 52 s (19.2 tx/s, 13 ledgers)** through the queue, no console errors, timestamps in order, against **462 over 33 s (13.9 tx/s)** on the bypass. The platforms no longer diverge: capacity, eviction counting and single-reader ordering hold on every target. One window remains and is not platform-specific - StartMessageProcessor runs after the OnConnected callback, so a handler subscribing there can see frames before the channel exists, and those take the direct fallback. Moving the start earlier is blocked by StartPingTimer calling StopPingTimerSync, which stops the processor too; tracked separately - * the issue's premise was checked and does not hold: handlers do not run in the receive loop. `ProcessStreamMessageAsync` is called by a background reader, and the receive loop only calls `TryWrite`. What is real is the silent loss, which is what this addresses + * the issue's premise was checked and does not hold: handlers do not run in the receive loop. A queued frame is dispatched by the background reader, and the receive loop does no more than `TryWrite`; a frame on the fallback path is handed off from the receive loop, but the hand-off yields before it parses anything, so the loop is not the thread that deserializes or calls handlers there either. What is real is the silent loss, which is what this addresses * **Frames from a socket being retired no longer reach handlers as if they were current** (#112). Session identity was threaded into every lifecycle callback - `OnceOpen`, `OnConnectionFailed`, `OnceClose` all compare against `_activeSession.SessionId` - and the message path was the one exception: `ws.OnBinaryMessage` called `IOnMessageFastPath(m)` with no session at all. * that mattered because retirement is not instant. `RetireOldSessionAsync` runs fire-and-forget beside the new connection and closes the old socket *gracefully*, so it keeps delivering for the length of the close handshake with its callback still attached. Whatever it sent landed in the new session's queue * after a reconnect those frames are stale. After a `ChangeServer` between networks they are worse than stale: switching mainnet to testnet, a handler that believes it is on testnet could receive the tail of mainnet's stream - transactions for accounts that do not exist there, ledger indexes from another chain, nothing marking them as belonging to the previous connection diff --git a/Tests/Xrpl.Tests/Client/TestUStaleSessionFrames.cs b/Tests/Xrpl.Tests/Client/TestUStaleSessionFrames.cs index e96de8e0..6de38f77 100644 --- a/Tests/Xrpl.Tests/Client/TestUStaleSessionFrames.cs +++ b/Tests/Xrpl.Tests/Client/TestUStaleSessionFrames.cs @@ -335,14 +335,60 @@ public async Task TestUFallbackPathReturnsBeforeHandlersRun() } /// - /// The guard must not reject frames from the live session, or the stream stops entirely. + /// A frame naming the live session is delivered. /// /// - /// Without this, a comparison that always failed would pass the test above while breaking - /// every subscription in the SDK. + /// The sessionless case below cannot stand in for this one: a guard that rejected every named + /// session would pass it and still take the whole stream down, since every frame the socket + /// produces is named. /// [TestMethod] - public async Task TestUFrameFromTheActiveSessionIsDelivered() + public async Task TestUFrameNamingTheLiveSessionIsDelivered() + { + using ScriptedResponseServer server = new ScriptedResponseServer(ScriptedReply); + using XrplClient client = new XrplClient(server.Url); + await client.Connect(); + await WaitForMessageProcessor(client); + + TaskCompletionSource received = new TaskCompletionSource( + TaskCreationOptions.RunContinuationsAsynchronously); + client.OnTransaction += r => + { + received.TrySetResult(r); + return Task.CompletedTask; + }; + + try + { + long? sessionId = client.connection.ActiveSessionId; + Assert.IsNotNull(sessionId, "a connected client must have a session"); + + await client.connection.IOnMessageFastPath( + Encoding.UTF8.GetBytes(TransactionMessage), sessionId); + + Task completed = await Task.WhenAny(received.Task, Task.Delay(TimeSpan.FromSeconds(5))); + Assert.AreSame(received.Task, completed, + "a frame from the session the client is actually on was dropped"); + + Assert.AreEqual(0L, client.connection.StaleSessionFramesDropped, + "nothing here came from a session the client had left"); + } + finally + { + await client.Disconnect(); + } + } + + /// + /// A frame that names no session is delivered. + /// + /// + /// OnMessage is public and has no session to name, so there is nothing to compare it + /// against. Rejecting it for want of a session would break the one entry point a caller can + /// drive by hand. + /// + [TestMethod] + public async Task TestUFrameNamingNoSessionIsDelivered() { using ScriptedResponseServer server = new ScriptedResponseServer(ScriptedReply); using XrplClient client = new XrplClient(server.Url); From 87c69be9d826e88119ac2255c08d560891a097b6 Mon Sep 17 00:00:00 2001 From: Aleksandr Platonenkov Date: Thu, 20 Aug 2026 13:27:16 -0300 Subject: [PATCH 10/16] =?UTF-8?q?test(client):=20=D0=BF=D1=80=D0=BE=D0=B2?= =?UTF-8?q?=D0=B5=D1=80=D0=BA=D0=B0=20=C2=AB=D0=BE=D0=B1=D1=80=D0=B0=D0=B1?= =?UTF-8?q?=D0=BE=D1=82=D1=87=D0=B8=D0=BA=20=D0=BD=D0=B5=20=D0=B2=D0=B8?= =?UTF-8?q?=D0=B4=D0=B5=D0=BB=20=D0=BA=D0=B0=D0=B4=D1=80=C2=BB=20=D0=BF?= =?UTF-8?q?=D0=B5=D1=80=D0=B5=D1=81=D1=82=D0=B0=D1=91=D1=82=20=D0=B1=D1=8B?= =?UTF-8?q?=D1=82=D1=8C=20=D0=B4=D0=B5=D0=BA=D0=BE=D1=80=D0=B0=D1=82=D0=B8?= =?UTF-8?q?=D0=B2=D0=BD=D0=BE=D0=B9?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit В двух тестах на отбрасывание стояло Assert.AreEqual(0, calls) сразу после инъекции. Упасть эта проверка не могла: выдача асинхронна на обоих путях, так что недоброшенный кадр просто ещё не пришёл бы к моменту проверки. Держал тесты только счётчик StaleSessionFramesDropped. Теперь после кадра под проверкой идёт свидетель — кадр без сессии через OnMessage, который не отвергается никогда, с отличимым Sequence. Он подан позже, а читатель один и разбирает очередь по порядку, поэтому если бы кадр под проверкой вообще попал в очередь, его выдали бы раньше свидетеля. Увидели свидетеля — значит второй не придёт уже никогда. Проверено мутацией, которую прежняя форма не ловила: считать кадр устаревшим, но всё равно доставить. Счётчик при этом равен единице и первая проверка проходит; краснеют обе проверки на свидетеля. --- .../Client/TestUStaleSessionFrames.cs | 62 ++++++++++++++++--- 1 file changed, 54 insertions(+), 8 deletions(-) diff --git a/Tests/Xrpl.Tests/Client/TestUStaleSessionFrames.cs b/Tests/Xrpl.Tests/Client/TestUStaleSessionFrames.cs index 6de38f77..102e349e 100644 --- a/Tests/Xrpl.Tests/Client/TestUStaleSessionFrames.cs +++ b/Tests/Xrpl.Tests/Client/TestUStaleSessionFrames.cs @@ -1,6 +1,8 @@ using Microsoft.VisualStudio.TestTools.UnitTesting; using System; +using System.Collections.Concurrent; +using System.Linq; using System.Text; using System.Threading; using System.Threading.Tasks; @@ -43,6 +45,42 @@ public class TestUStaleSessionFrames } """; + /// + /// The same message with a sequence number a test can recognise. + /// + private static string TransactionMessageWith(uint sequence) => + TransactionMessage.Replace("\"Sequence\": 9", $"\"Sequence\": {sequence}"); + + private const uint DroppedSequence = 9; + + private const uint WitnessSequence = 77; + + /// + /// Waits until the handler has seen the witness frame, then reports whether it also saw the + /// frame that was supposed to be dropped. + /// + /// + /// A bare "the handler was not called" assertion cannot fail: dispatch is asynchronous on both + /// paths, so an undropped frame would simply not have arrived yet when the assertion ran. The + /// witness fixes that. It carries no session - OnMessage, which is never rejected - and + /// is submitted after the frame under test, so the single reader would have dispatched that + /// one first had it been queued at all. Seeing the witness therefore means the other one is + /// never coming. + /// + private static async Task DriveWitnessAndWait(XrplClient client, ConcurrentQueue seen) + { + await client.connection.OnMessage(TransactionMessageWith(WitnessSequence)); + + DateTime deadline = DateTime.UtcNow.AddSeconds(5); + while (!seen.Contains(WitnessSequence) && DateTime.UtcNow < deadline) + { + await Task.Delay(10); + } + + Assert.IsTrue(seen.Contains(WitnessSequence), + "the witness frame never arrived, so nothing here can be concluded about the other one"); + } + /// /// Waits until stream frames are queued rather than taking the fallback path. /// @@ -72,11 +110,12 @@ public async Task TestUFrameFromARetiredSessionNeverReachesHandlers() using ScriptedResponseServer server = new ScriptedResponseServer(ScriptedReply); using XrplClient client = new XrplClient(server.Url); await client.Connect(); + await WaitForMessageProcessor(client); - int calls = 0; - client.OnTransaction += _ => + ConcurrentQueue seen = new ConcurrentQueue(); + client.OnTransaction += r => { - calls++; + seen.Enqueue(r.Transaction.Sequence ?? 0); return Task.CompletedTask; }; @@ -90,7 +129,10 @@ await client.connection.IOnMessageFastPath( Assert.AreEqual(1L, client.connection.StaleSessionFramesDropped, "a frame from a session that is not active must be dropped before it reaches the queue"); - Assert.AreEqual(0, calls, "the handler saw a frame belonging to a connection that is being retired"); + + await DriveWitnessAndWait(client, seen); + Assert.IsFalse(seen.Contains(DroppedSequence), + "the handler saw a frame belonging to a connection that is being retired"); } finally { @@ -114,11 +156,12 @@ public async Task TestUFrameFromTheSessionBeingRetiredIsDropped() using ScriptedResponseServer server = new ScriptedResponseServer(ScriptedReply); using XrplClient client = new XrplClient(server.Url); await client.Connect(); + await WaitForMessageProcessor(client); - int calls = 0; - client.OnTransaction += _ => + ConcurrentQueue seen = new ConcurrentQueue(); + client.OnTransaction += r => { - calls++; + seen.Enqueue(r.Transaction.Sequence ?? 0); return Task.CompletedTask; }; @@ -134,7 +177,10 @@ await client.connection.IOnMessageFastPath( Assert.AreEqual(1L, client.connection.StaleSessionFramesDropped, "an id that matches a retiring session is not an id that matches the live one"); - Assert.AreEqual(0, calls, "the handler saw a frame from the session being retired"); + + await DriveWitnessAndWait(client, seen); + Assert.IsFalse(seen.Contains(DroppedSequence), + "the handler saw a frame from the session being retired"); } finally { From 92310d97680557a1222dbdde814b29d6e5eaad1b Mon Sep 17 00:00:00 2001 From: Aleksandr Platonenkov Date: Thu, 20 Aug 2026 14:00:27 -0300 Subject: [PATCH 11/16] =?UTF-8?q?test(client):=20=D1=88=D0=BE=D0=B2=20?= =?UTF-8?q?=D0=BF=D0=BE=D0=BC=D0=B5=D1=87=D0=B0=D0=B5=D1=82=20=D1=81=D0=B5?= =?UTF-8?q?=D1=81=D1=81=D0=B8=D1=8E=20=D0=B8=20=D0=BD=D0=B0=D0=B7=D1=8B?= =?UTF-8?q?=D0=B2=D0=B0=D0=B5=D1=82=20=D0=B5=D1=91=20=D0=BE=D0=B4=D0=BD?= =?UTF-8?q?=D0=B8=D0=BC=20=D0=B7=D0=B0=D0=BC=D0=BA=D0=BE=D0=BC?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit MarkActiveSessionRetiringForTests() снова возвращает id помеченной сессии. Разделение на ActiveSessionId + пометку я сделал сам во время селф-ревью «ради чистоты» и внёс этим TOCTOU: два захвата замка позволяют переподключению подменить сессию между ними, и тест, назвавший прочитанный первым id, проверял бы уже путь несовпадения — тот самый, который держит соседний тест, — молча и только иногда. ActiveSessionId остаётся: им пользуются тесты, которые ничего не помечают, и там разрыва нет. Проверено мутацией: снятие клаузулы !IsRetiring по-прежнему красит этот тест. --- Tests/Xrpl.Tests/Client/TestUStaleSessionFrames.cs | 5 +++-- Xrpl/Client/connection.cs | 12 +++++++++++- 2 files changed, 14 insertions(+), 3 deletions(-) diff --git a/Tests/Xrpl.Tests/Client/TestUStaleSessionFrames.cs b/Tests/Xrpl.Tests/Client/TestUStaleSessionFrames.cs index 102e349e..08411da2 100644 --- a/Tests/Xrpl.Tests/Client/TestUStaleSessionFrames.cs +++ b/Tests/Xrpl.Tests/Client/TestUStaleSessionFrames.cs @@ -167,9 +167,10 @@ public async Task TestUFrameFromTheSessionBeingRetiredIsDropped() try { - long? retiringSessionId = client.connection.ActiveSessionId; + // Marked and named in one lock: reading the id separately would let a reconnect swap + // the session in between, and the test would quietly fall back to the mismatch case. + long? retiringSessionId = client.connection.MarkActiveSessionRetiringForTests(); Assert.IsNotNull(retiringSessionId, "a connected client must have a session to retire"); - client.connection.MarkActiveSessionRetiringForTests(); await client.connection.IOnMessageFastPath( Encoding.UTF8.GetBytes(TransactionMessage), diff --git a/Xrpl/Client/connection.cs b/Xrpl/Client/connection.cs index e1558973..08981022 100644 --- a/Xrpl/Client/connection.cs +++ b/Xrpl/Client/connection.cs @@ -497,12 +497,22 @@ internal void CompleteStreamChannelWriterForTests() /// /// Exists for tests: in production that window is reachable only by racing a real reconnect. /// Marks it exactly the way the two production paths do, under _sessionLock. + /// + /// Returns the id rather than leaving the caller to read + /// separately: two lock acquisitions would let a reconnect swap the session in between, and a + /// test that then named the id it read first would be exercising the mismatch path it was + /// written to avoid - silently, and only sometimes. + /// /// - internal void MarkActiveSessionRetiringForTests() + /// + /// The id of the session that was marked, or if there is none. + /// + internal long? MarkActiveSessionRetiringForTests() { lock (_sessionLock) { _activeSession?.MarkAsRetiring(); + return _activeSession?.SessionId; } } From fa25ad0a60a9f41e4df0da16c96b3cb21ec8da93 Mon Sep 17 00:00:00 2001 From: Aleksandr Platonenkov Date: Thu, 20 Aug 2026 14:25:08 -0300 Subject: [PATCH 12/16] =?UTF-8?q?feat(client):=20=D1=81=D1=87=D1=91=D1=82?= =?UTF-8?q?=D1=87=D0=B8=D0=BA=20=D0=BA=D0=B0=D0=B4=D1=80=D0=BE=D0=B2=20?= =?UTF-8?q?=D0=BC=D0=B8=D0=BC=D0=BE=20=D0=BE=D1=87=D0=B5=D1=80=D0=B5=D0=B4?= =?UTF-8?q?=D0=B8=20=D0=B8=20=D1=82=D0=B5=D0=BB=D0=B0=20=D0=BF=D0=BE=20?= =?UTF-8?q?=D1=83=D0=BC=D0=BE=D0=BB=D1=87=D0=B0=D0=BD=D0=B8=D1=8E=20=D0=B2?= =?UTF-8?q?=20=D0=B8=D0=BD=D1=82=D0=B5=D1=80=D1=84=D0=B5=D0=B9=D1=81=D0=B5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit FallbackDispatchedStreamMessages считает кадры, выданные в обход очереди. У запасного пути нет ни границы ёмкости, ни учёта вытеснения, ни порядка от единственного читателя, а снаружи не было видно ни того, что он вообще использовался, ни насколько. Уводят туда три вещи: процессор ещё не поднят, процессор остановлен, канал отказал в записи. Первое — реальное окно на каждом подключении, и теперь оно измеримо, а не только обсуждаемо. Все три счётчика объявлены в IXrplClient с телами по умолчанию, проброс в connection — единственная реализация, которая имеет смысл. DroppedStreamMessages был объявлен без тела в этом же невыпущенном цикле; дать тело и ему ничего не стоит и сохраняет компилируемость внешней реализации интерфейса. Прецедент в файле есть — SetNetworkId. Проверено мутацией: снятие инкремента красит тест на отказ канала. --- CHANGES.md | 3 +++ .../Client/TestUStaleSessionFrames.cs | 3 +++ Xrpl/Client/IXrplClient.cs | 26 +++++++++++++++++-- Xrpl/Client/connection.cs | 21 +++++++++++++++ 4 files changed, 51 insertions(+), 2 deletions(-) diff --git a/CHANGES.md b/CHANGES.md index b17ef76e..f3d2b2a2 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -21,6 +21,7 @@ This release makes the SDK stop misrepresenting what a node sent. It is a delibe | `IXrplClient.connection` | `{ get; set; }` | `{ get; }` — assigning it would strand handlers on the old object | | Dropped stream events | invisible | `client.DroppedStreamMessages` counts them; `StreamMessageQueueCapacity` sizes the queue | | Frames from a retiring socket | delivered as current | dropped, and counted by `Connection.StaleSessionFramesDropped` | +| Frames dispatched outside the queue | invisible | counted by `Connection.FallbackDispatchedStreamMessages` | | 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 | @@ -213,6 +214,8 @@ The entries below are grouped by what changed, not by the order the levels were * matching the id alone would not do either: both paths call `MarkAsRetiring()` while the session is still the active one and only `ConnectInternalAsync` installs its replacement, so frames arriving in that window carry an id that matches. The retiring flag is part of the test, under `_sessionLock` - the same guard `OnceOpen` and the other lifecycle callbacks already use, and the only way `IsRetiring` is published at all * a `null` session means the caller has none to name (`OnMessage`, which anyone may call), and nothing is rejected in that case * consequence worth naming: `OnMessage` no longer dispatches inline when there is no processor. It never did on the queued path, so this makes the two agree - but code that called `OnMessage` on an unconnected client and read handler state on the next line was relying on the difference + * **what goes round the queue is now countable.** `FallbackDispatchedStreamMessages` counts frames dispatched outside it - no capacity bound, no eviction counting, no single-reader ordering apply to them. Three things send a frame there: the processor not being up yet, the processor having been stopped, and a refused write. The first is a real window on every connect, and this turns it from something argued about into something measured + * all three counters are declared with default bodies on `IXrplClient`, forwarding to `connection` - the only implementation that means anything. `DroppedStreamMessages` was declared without one in this same unreleased cycle; giving it one too costs nothing and keeps an external implementation of the interface compiling * **the fallback path hands the frame off before doing any work.** An async method runs on its caller's thread up to the first real await, and the first real await inside `ProcessStreamMessageAsync` comes after `JsonSerializer.Deserialize` - so every frame taking the fallback had its JSON parsed on the receive loop, plus whatever a handler did before its own first await. That is the head-of-line blocking the queue exists to prevent, reintroduced for the startup window, for a stopped processor and for a refused write. A yield at the top of the fallback ends it * **a frame the channel refuses no longer vanishes.** `TryWrite` was called for its side effect and its result ignored, on the reasoning that a `DropOldest` channel never refuses - which is true of a full queue (it evicts, counts through `itemDropped` and reports success) and false of a completed one. `StopMessageProcessorInternal` completes the writer *after* clearing `_streamMessageChannel`, so whoever read the reference an instant earlier writes into a closed channel and the frame was dropped with nothing to show for it. Not a corner case: `StartPingTimer` tears the processor down and `StartMessageProcessor` builds it again on every connect. Such a frame now takes the fallback path, where it still faces the session check * `Connection.StaleSessionFramesDropped` counts what was discarded, kept separate from `DroppedStreamMessages` because the two mean different things: a non-zero value here is normal right after a reconnect, while the other means consumers are falling behind. Also on `IXrplClient`, next to `DroppedStreamMessages` - a counter nobody can read is not observability. **New interface member**: an external implementation of `IXrplClient` has to add it diff --git a/Tests/Xrpl.Tests/Client/TestUStaleSessionFrames.cs b/Tests/Xrpl.Tests/Client/TestUStaleSessionFrames.cs index 08411da2..478aa780 100644 --- a/Tests/Xrpl.Tests/Client/TestUStaleSessionFrames.cs +++ b/Tests/Xrpl.Tests/Client/TestUStaleSessionFrames.cs @@ -313,6 +313,9 @@ await client.connection.IOnMessageFastPath( Assert.AreEqual(0L, client.connection.StaleSessionFramesDropped, "the session was live throughout; nothing here is stale"); + + Assert.AreEqual(1L, client.connection.FallbackDispatchedStreamMessages, + "a frame that went round the queue must say so - that is what the counter is for"); } finally { diff --git a/Xrpl/Client/IXrplClient.cs b/Xrpl/Client/IXrplClient.cs index 649cc4a9..0050a0f1 100644 --- a/Xrpl/Client/IXrplClient.cs +++ b/Xrpl/Client/IXrplClient.cs @@ -67,7 +67,12 @@ public interface IXrplClient : IDisposable /// socket - silently, until something reads this. A consumer building state from the /// stream should treat any increase as a signal that its state has drifted. /// - long DroppedStreamMessages { get; } + /// + /// Defaulted, like : forwarding to is + /// the only implementation that means anything, and an observational member is a poor + /// reason to break every external implementer of this interface. + /// + long DroppedStreamMessages => connection.DroppedStreamMessages; /// /// How many stream frames were discarded because they came from a connection this client @@ -81,7 +86,21 @@ public interface IXrplClient : IDisposable /// reconnect and says nothing about handler speed - that is /// , and the two are counted apart on purpose. /// - long StaleSessionFramesDropped { get; } + /// + long StaleSessionFramesDropped => connection.StaleSessionFramesDropped; + + /// + /// How many stream frames were dispatched outside the queue, without its ordering or its + /// capacity bound. + /// + /// + /// Frames take that path when the background processor is not up yet, when it has been + /// stopped, or when the channel refuses a write. The first of those is a real window on + /// every connect: the processor starts at the end of OnceOpen, after the + /// OnConnected callback, so a handler subscribing there can be reached before the + /// queue exists. This counts how often that happened. + /// + long FallbackDispatchedStreamMessages => connection.FallbackDispatchedStreamMessages; /// Node error reported over the socket. event OnError OnError; @@ -576,6 +595,9 @@ public class ClientOptions : ConnectionOptions /// public long StaleSessionFramesDropped => connection.StaleSessionFramesDropped; + /// + public long FallbackDispatchedStreamMessages => connection.FallbackDispatchedStreamMessages; + // Forwarded, not relayed: add/remove reach the same Connection a caller would have used // through the property, so this type holds no delegates and no subscription of its own. A // relaying version - a local event plus a subscription to the connection that re-raises it diff --git a/Xrpl/Client/connection.cs b/Xrpl/Client/connection.cs index 08981022..5f386424 100644 --- a/Xrpl/Client/connection.cs +++ b/Xrpl/Client/connection.cs @@ -425,6 +425,26 @@ private static WebSocketClient CreateWebSocket(string url, ConnectionOptions con private long _staleSessionFramesDropped; + private long _fallbackDispatchedStreamMessages; + + /// + /// How many stream frames were dispatched outside the queue. + /// + /// + /// The fallback path holds none of the queue's guarantees: no capacity bound, no eviction + /// counting, no single-reader ordering. Three things send a frame down it - the processor not + /// being up yet, the processor having been stopped, and the channel refusing a write because + /// 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. + /// + /// + public long FallbackDispatchedStreamMessages => Interlocked.Read(ref _fallbackDispatchedStreamMessages); + /// /// How many stream frames were discarded because they came from a session that is no longer /// active. @@ -3617,6 +3637,7 @@ private void EnqueueStreamMessage(byte[] frame, long? sessionId = null) return; } + Interlocked.Increment(ref _fallbackDispatchedStreamMessages); _ = ProcessStreamMessageFireAndForgetAsync(item); } From 3ded7b5f077cfab56982d3eb2e3921fd9adbbdec Mon Sep 17 00:00:00 2001 From: Aleksandr Platonenkov Date: Thu, 20 Aug 2026 15:05:46 -0300 Subject: [PATCH 13/16] =?UTF-8?q?docs(client):=20=D0=B1=D0=B8=D1=82=D1=8B?= =?UTF-8?q?=D0=B5=20crefs=20=D0=B8=20=D0=BF=D1=80=D0=BE=D1=82=D0=B8=D0=B2?= =?UTF-8?q?=D0=BE=D1=80=D0=B5=D1=87=D0=B8=D0=B5=20=D0=B2=20CHANGES?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit CHANGES утверждал, что внешней реализации IXrplClient придётся добавить новый член, — это осталось от версии до тел по умолчанию и противоречило соседнему пункту в том же списке. Два cref не разрешались и давали CS1574 при генерации документации: IOnMessageFastPath(string, byte[]) — сигнатура, которой больше нет, и Volatile.Read{T}(ref T), который я написал в прошлом коммите. Проверено сборкой: с ними 8 предупреждений CS1574, без них — 2, и обе оставшиеся в файлах, которых этот PR не касается (BaseResponse.cs, BookChangesStream.cs). У DroppedStreamMessages было два блока подряд, у StaleSessionFramesDropped — собственные summary и remarks плюс inheritdoc на соседа, то есть два конфликтующих источника документации для одного члена. Оба следа от прошлого коммита: заметка про тело по умолчанию теперь внутри собственных remarks каждого из трёх счётчиков. FeeTestClient объявляет и FallbackDispatchedStreamMessages: этот двойник возвращает null из connection, а тела по умолчанию пробрасывают именно туда, так что унаследованный член кинул бы NullReferenceException при первом чтении. --- CHANGES.md | 2 +- Tests/Xrpl.Tests/Sugar/TestUAutofillFees.cs | 4 ++++ Xrpl/Client/IXrplClient.cs | 15 ++++++++++++--- Xrpl/Client/connection.cs | 4 ++-- 4 files changed, 19 insertions(+), 6 deletions(-) diff --git a/CHANGES.md b/CHANGES.md index f3d2b2a2..0c59efcb 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -218,7 +218,7 @@ The entries below are grouped by what changed, not by the order the levels were * all three counters are declared with default bodies on `IXrplClient`, forwarding to `connection` - the only implementation that means anything. `DroppedStreamMessages` was declared without one in this same unreleased cycle; giving it one too costs nothing and keeps an external implementation of the interface compiling * **the fallback path hands the frame off before doing any work.** An async method runs on its caller's thread up to the first real await, and the first real await inside `ProcessStreamMessageAsync` comes after `JsonSerializer.Deserialize` - so every frame taking the fallback had its JSON parsed on the receive loop, plus whatever a handler did before its own first await. That is the head-of-line blocking the queue exists to prevent, reintroduced for the startup window, for a stopped processor and for a refused write. A yield at the top of the fallback ends it * **a frame the channel refuses no longer vanishes.** `TryWrite` was called for its side effect and its result ignored, on the reasoning that a `DropOldest` channel never refuses - which is true of a full queue (it evicts, counts through `itemDropped` and reports success) and false of a completed one. `StopMessageProcessorInternal` completes the writer *after* clearing `_streamMessageChannel`, so whoever read the reference an instant earlier writes into a closed channel and the frame was dropped with nothing to show for it. Not a corner case: `StartPingTimer` tears the processor down and `StartMessageProcessor` builds it again on every connect. Such a frame now takes the fallback path, where it still faces the session check - * `Connection.StaleSessionFramesDropped` counts what was discarded, kept separate from `DroppedStreamMessages` because the two mean different things: a non-zero value here is normal right after a reconnect, while the other means consumers are falling behind. Also on `IXrplClient`, next to `DroppedStreamMessages` - a counter nobody can read is not observability. **New interface member**: an external implementation of `IXrplClient` has to add it + * `Connection.StaleSessionFramesDropped` counts what was discarded, kept separate from `DroppedStreamMessages` because the two mean different things: a non-zero value here is normal right after a reconnect, while the other means consumers are falling behind. Also on `IXrplClient`, next to `DroppedStreamMessages` - a counter nobody can read is not observability. **New interface member, with a default body** forwarding to `connection`: an external implementation of `IXrplClient` keeps compiling and may override it ## 10.12.0.0 08/16/2026 diff --git a/Tests/Xrpl.Tests/Sugar/TestUAutofillFees.cs b/Tests/Xrpl.Tests/Sugar/TestUAutofillFees.cs index 48a9c722..942dd377 100644 --- a/Tests/Xrpl.Tests/Sugar/TestUAutofillFees.cs +++ b/Tests/Xrpl.Tests/Sugar/TestUAutofillFees.cs @@ -645,6 +645,10 @@ public FeeTestClient(string feeXrp, uint reserveInc, string maxFeeXRP = "5") public long StaleSessionFramesDropped => 0; + // Declared rather than inherited: this substitute returns null from connection, and the + // interface defaults forward there. + public long FallbackDispatchedStreamMessages => 0; + // Declared, never raised: this client answers requests, it does not stream. Being able to // declare them at all is the point of the events moving onto IXrplClient - a substitute // client could not carry them while they lived only on Connection. diff --git a/Xrpl/Client/IXrplClient.cs b/Xrpl/Client/IXrplClient.cs index 0050a0f1..d0a03128 100644 --- a/Xrpl/Client/IXrplClient.cs +++ b/Xrpl/Client/IXrplClient.cs @@ -66,11 +66,11 @@ public interface IXrplClient : IDisposable /// and drops the oldest when full, so a slow handler costs events instead of stalling the /// socket - silently, until something reads this. A consumer building state from the /// stream should treat any increase as a signal that its state has drifted. - /// - /// + /// /// Defaulted, like : forwarding to is /// the only implementation that means anything, and an observational member is a poor /// reason to break every external implementer of this interface. + /// /// long DroppedStreamMessages => connection.DroppedStreamMessages; @@ -85,8 +85,12 @@ public interface IXrplClient : IDisposable /// different chain entirely. A non-zero value here is therefore normal right after a /// reconnect and says nothing about handler speed - that is /// , and the two are counted apart on purpose. + /// + /// Defaulted, like : forwarding to is + /// the only implementation that means anything, and an observational member is a poor + /// reason to break every external implementer of this interface. + /// /// - /// long StaleSessionFramesDropped => connection.StaleSessionFramesDropped; /// @@ -99,6 +103,11 @@ public interface IXrplClient : IDisposable /// every connect: the processor starts at the end of OnceOpen, after the /// OnConnected callback, so a handler subscribing there can be reached before the /// queue exists. This counts how often that happened. + /// + /// Defaulted, like : forwarding to is + /// the only implementation that means anything, and an observational member is a poor + /// reason to break every external implementer of this interface. + /// /// long FallbackDispatchedStreamMessages => connection.FallbackDispatchedStreamMessages; diff --git a/Xrpl/Client/connection.cs b/Xrpl/Client/connection.cs index 5f386424..4902fee6 100644 --- a/Xrpl/Client/connection.cs +++ b/Xrpl/Client/connection.cs @@ -489,7 +489,7 @@ internal long? ActiveSessionId /// . /// /// - /// rather than a plain read or _messageProcessorLock: + /// 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. /// @@ -3360,7 +3360,7 @@ private Task IOnMessageFastPath(string message) /// /// Overload for a message still in its wire form, used by the socket callback. See - /// for why the bytes are kept as they are. + /// for why the bytes are kept as they are. /// /// /// Internal rather than private so a test can drive the actual production entry point - the From a69eb39d6db8cfcb6b48f535eab44c4725d4e5fd Mon Sep 17 00:00:00 2001 From: Aleksandr Platonenkov Date: Thu, 20 Aug 2026 15:13:11 -0300 Subject: [PATCH 14/16] =?UTF-8?q?docs:=20=D1=81=D1=82=D1=80=D0=BE=D0=BA?= =?UTF-8?q?=D0=B0=20=D1=82=D0=B0=D0=B1=D0=BB=D0=B8=D1=86=D1=8B=20=D0=BE?= =?UTF-8?q?=D0=BF=D0=B8=D1=81=D1=8B=D0=B2=D0=B0=D0=BB=D0=B0=20=D1=82=D0=BE?= =?UTF-8?q?=D0=BB=D1=8C=D0=BA=D0=BE=20=D0=BF=D0=BE=D0=BB=D0=BE=D0=B2=D0=B8?= =?UTF-8?q?=D0=BD=D1=83=20=D1=81=D0=BB=D1=83=D1=87=D0=B0=D0=B5=D0=B2?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit StaleSessionFramesDropped считает кадры двух видов: с id сессии, которой уже нет вовсе, и с id сессии, которую сейчас хоронят, — она ещё активна, и именно поэтому для неё понадобилась клаузула !IsRetiring. «Frames from a retiring socket» называло только второй. --- CHANGES.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/CHANGES.md b/CHANGES.md index 0c59efcb..dc9b677f 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -20,7 +20,7 @@ This release makes the SDK stop misrepresenting what a node sent. It is a delibe | Stream events | `client.connection.OnTransaction += …` | also `client.OnTransaction += …` — on `IXrplClient` now; the old form still works | | `IXrplClient.connection` | `{ get; set; }` | `{ get; }` — assigning it would strand handlers on the old object | | Dropped stream events | invisible | `client.DroppedStreamMessages` counts them; `StreamMessageQueueCapacity` sizes the queue | -| Frames from a retiring socket | delivered as current | dropped, and counted by `Connection.StaleSessionFramesDropped` | +| 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` | | 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 | From b818d697d4ee05844f3bb89bb64c126b470329dd Mon Sep 17 00:00:00 2001 From: Aleksandr Platonenkov Date: Thu, 20 Aug 2026 15:23:07 -0300 Subject: [PATCH 15/16] =?UTF-8?q?docs:=20=D0=B3=D0=B0=D1=80=D0=B0=D0=BD?= =?UTF-8?q?=D1=82=D0=B8=D0=B8=20=D0=BE=D1=87=D0=B5=D1=80=D0=B5=D0=B4=D0=B8?= =?UTF-8?q?=20=D0=BD=D0=B5=20=D1=80=D0=B0=D1=81=D0=BF=D1=80=D0=BE=D1=81?= =?UTF-8?q?=D1=82=D1=80=D0=B0=D0=BD=D1=8F=D1=8E=D1=82=D1=81=D1=8F=20=D0=BD?= =?UTF-8?q?=D0=B0=20=D0=B7=D0=B0=D0=BF=D0=B0=D1=81=D0=BD=D0=BE=D0=B9=20?= =?UTF-8?q?=D0=BF=D1=83=D1=82=D1=8C,=20=D0=B0=20yield=20=E2=80=94=20=D0=BD?= =?UTF-8?q?=D0=B5=20=D0=BF=D1=80=D0=BE=20=D0=BF=D0=BE=D1=82=D0=BE=D0=BA?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Две неточности в формулировках, обе меняют смысл. «Capacity, eviction counting and single-reader ordering hold on every target» звучало как безусловное, хотя четырьмя пунктами ниже сказано ровно обратное про кадры на запасном пути. Оговорено: на пути через очередь и для кадров, которые в неё попали. «The loop is not the thread that deserializes» — Task.Yield() гарантирует асинхронность продолжения, а не другой поток операционной системы; под однопоточным WebAssembly это тот же поток, просто позже. Утверждение переформулировано на «не делает этого синхронно» — а больше здесь и не нужно. То же исправлено в doc-комментарии теста, где стояла та же фраза. --- CHANGES.md | 4 ++-- Tests/Xrpl.Tests/Client/TestUStaleSessionFrames.cs | 4 ++-- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/CHANGES.md b/CHANGES.md index dc9b677f..74392012 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -205,8 +205,8 @@ The entries below are grouped by what changed, not by the order the levels were * `Connection.DroppedStreamMessages` and `IXrplClient.DroppedStreamMessages` count the discards, across reconnects and `ChangeServer` alike, since one `Connection` serves them all. Any increase means events arrived and never reached a handler * `ConnectionOptions.StreamMessageQueueCapacity` (default 10 000, unchanged) sizes the queue - raise it for a consumer that must not miss events and can hold the frames, lower it to bound memory harder * the counter is incremented from `Channel`'s `itemDropped` callback, which runs inside `TryWrite` on the receive loop. It does nothing but increment for that reason: raising an event or logging there would put consumer code back on the path this queue exists to keep it off - the very failure #105 described but which the queue already prevented - * **browsers no longer take a separate path** (#110). `EnqueueStreamMessage` used to start one fire-and-forget task per frame under WebAssembly, bypassing the queue: the capacity was not consulted, nothing was evicted, the counter stayed at zero however far handlers fell behind, the backlog was bounded by nothing, and concurrent dispatch could hand handlers events out of the order the node sent them. The queue was built for that environment to begin with - `StartMessageProcessor` says "true async support in WebAssembly single-threaded environment" - and measurement confirms it works there. Running the Blazor demo against mainnet: **1 004 transactions over 52 s (19.2 tx/s, 13 ledgers)** through the queue, no console errors, timestamps in order, against **462 over 33 s (13.9 tx/s)** on the bypass. The platforms no longer diverge: capacity, eviction counting and single-reader ordering hold on every target. One window remains and is not platform-specific - StartMessageProcessor runs after the OnConnected callback, so a handler subscribing there can see frames before the channel exists, and those take the direct fallback. Moving the start earlier is blocked by StartPingTimer calling StopPingTimerSync, which stops the processor too; tracked separately - * the issue's premise was checked and does not hold: handlers do not run in the receive loop. A queued frame is dispatched by the background reader, and the receive loop does no more than `TryWrite`; a frame on the fallback path is handed off from the receive loop, but the hand-off yields before it parses anything, so the loop is not the thread that deserializes or calls handlers there either. What is real is the silent loss, which is what this addresses + * **browsers no longer take a separate path** (#110). `EnqueueStreamMessage` used to start one fire-and-forget task per frame under WebAssembly, bypassing the queue: the capacity was not consulted, nothing was evicted, the counter stayed at zero however far handlers fell behind, the backlog was bounded by nothing, and concurrent dispatch could hand handlers events out of the order the node sent them. The queue was built for that environment to begin with - `StartMessageProcessor` says "true async support in WebAssembly single-threaded environment" - and measurement confirms it works there. Running the Blazor demo against mainnet: **1 004 transactions over 52 s (19.2 tx/s, 13 ledgers)** through the queue, no console errors, timestamps in order, against **462 over 33 s (13.9 tx/s)** on the bypass. The platforms no longer diverge on the queued path: capacity, eviction counting and single-reader ordering hold on every target, for frames that enter the queue. One window remains and is not platform-specific - StartMessageProcessor runs after the OnConnected callback, so a handler subscribing there can see frames before the channel exists, and those take the direct fallback. Moving the start earlier is blocked by StartPingTimer calling StopPingTimerSync, which stops the processor too; tracked separately + * the issue's premise was checked and does not hold: handlers do not run in the receive loop. A queued frame is dispatched by the background reader, and the receive loop does no more than `TryWrite`; a frame on the fallback path is handed off from the receive loop, but the hand-off yields before it parses anything, so the receive loop does not synchronously deserialize or call handlers there either - the yield promises asynchrony, not a different thread, which is the whole promise needed here. What is real is the silent loss, which is what this addresses * **Frames from a socket being retired no longer reach handlers as if they were current** (#112). Session identity was threaded into every lifecycle callback - `OnceOpen`, `OnConnectionFailed`, `OnceClose` all compare against `_activeSession.SessionId` - and the message path was the one exception: `ws.OnBinaryMessage` called `IOnMessageFastPath(m)` with no session at all. * that mattered because retirement is not instant. `RetireOldSessionAsync` runs fire-and-forget beside the new connection and closes the old socket *gracefully*, so it keeps delivering for the length of the close handshake with its callback still attached. Whatever it sent landed in the new session's queue * after a reconnect those frames are stale. After a `ChangeServer` between networks they are worse than stale: switching mainnet to testnet, a handler that believes it is on testnet could receive the tail of mainnet's stream - transactions for accounts that do not exist there, ledger indexes from another chain, nothing marking them as belonging to the previous connection diff --git a/Tests/Xrpl.Tests/Client/TestUStaleSessionFrames.cs b/Tests/Xrpl.Tests/Client/TestUStaleSessionFrames.cs index 478aa780..3e294e08 100644 --- a/Tests/Xrpl.Tests/Client/TestUStaleSessionFrames.cs +++ b/Tests/Xrpl.Tests/Client/TestUStaleSessionFrames.cs @@ -324,8 +324,8 @@ await client.connection.IOnMessageFastPath( } /// - /// The fallback path hands the frame off before doing any of the work, so the receive loop is - /// not the thread that parses JSON and runs handlers. + /// The fallback path hands the frame off before doing any of the work, so the receive loop + /// does not synchronously parse JSON or run handlers. /// /// /// An async method runs on its caller's thread up to the first real await, and the first real From 8a22fc199c6e291d1662e858d5afbbf97f36ba34 Mon Sep 17 00:00:00 2001 From: Aleksandr Platonenkov Date: Thu, 20 Aug 2026 16:10:58 -0300 Subject: [PATCH 16/16] =?UTF-8?q?test(client):=20=D1=81=D0=B2=D0=B8=D0=B4?= =?UTF-8?q?=D0=B5=D1=82=D0=B5=D0=BB=D1=8C=20=D1=80=D0=B5=D0=B3=D0=B8=D1=81?= =?UTF-8?q?=D1=82=D1=80=D0=B8=D1=80=D1=83=D0=B5=D1=82=D1=81=D1=8F=20=D0=BF?= =?UTF-8?q?=D0=BE=D1=81=D0=BB=D0=B5=20=D0=BE=D0=B1=D1=80=D0=B0=D0=B1=D0=BE?= =?UTF-8?q?=D1=82=D1=87=D0=B8=D0=BA=D0=B0,=20=D1=81=D1=87=D1=91=D1=82?= =?UTF-8?q?=D1=87=D0=B8=D0=BA=20=D0=BF=D1=80=D0=BE=D0=B2=D0=B5=D1=80=D1=8F?= =?UTF-8?q?=D0=B5=D1=82=D1=81=D1=8F=20=D0=B4=D0=B5=D0=BB=D1=8C=D1=82=D0=BE?= =?UTF-8?q?=D0=B9?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit В первом тесте форвардинга комментарий утверждал, что свидетель добавлен после обработчика, а код добавлял его до. Многоадресный делегат вызывает подписчиков в порядке регистрации, так что свидетель отчитывался о выдаче кадра раньше, чем отрабатывал обработчик, — проверка «sanity: the handler is attached» могла мигать, а не падать. Порядок исправлен, и почему он важен — теперь написано в обоих тестах, чтобы его не переставили обратно из соображений опрятности. FallbackDispatchedStreamMessages проверяется дельтой от базовой линии: счётчик живёт всё соединение, а стартовое окно, описанное в этом же файле, могло увести кадры мимо очереди до того, как тест до него добрался. Проверено мутацией: замена remove на += в аксессоре клиента по-прежнему красит зеркальный тест. --- .../Client/TestUClientStreamEvents.cs | 17 +++++++++++++---- .../Client/TestUStaleSessionFrames.cs | 7 ++++++- 2 files changed, 19 insertions(+), 5 deletions(-) diff --git a/Tests/Xrpl.Tests/Client/TestUClientStreamEvents.cs b/Tests/Xrpl.Tests/Client/TestUClientStreamEvents.cs index a154e3d5..302ce755 100644 --- a/Tests/Xrpl.Tests/Client/TestUClientStreamEvents.cs +++ b/Tests/Xrpl.Tests/Client/TestUClientStreamEvents.cs @@ -112,9 +112,16 @@ public async Task TestUHandlerAddedThroughTheClientCanBeRemovedThroughTheConnect // that work: a relaying client would keep its own subscriber list, the removal would miss // it, and the handler would keep firing. Removing through the same surface it was added // to cannot tell the two apart - both pass - which is why the test crosses surfaces. - // A witness that is never removed, added after the handler so it runs after it: it tells - // the test when a frame has been dispatched. Without it, "calls is still 1" after the - // removal would pass just as well for a frame that has not been processed yet. + contract.OnTransaction += handler; + + // A witness that is never removed: it tells the test when a frame has been dispatched. + // Without it, "calls is still 1" after the removal would pass just as well for a frame + // that has not been processed yet. + // + // Registered after the handler, and that order is load-bearing: a multicast delegate + // invokes its subscribers in registration order, so a witness registered first would + // report the frame as dispatched while the handler had not yet run - and the sanity + // assertion below would flake rather than fail. int dispatched = 0; client.connection.OnTransaction += _ => { @@ -122,7 +129,6 @@ public async Task TestUHandlerAddedThroughTheClientCanBeRemovedThroughTheConnect return Task.CompletedTask; }; - contract.OnTransaction += handler; await client.connection.OnMessage(TransactionMessage); WaitForCount(() => Volatile.Read(ref dispatched), 1, "the first frame was never dispatched"); Assert.AreEqual(1, calls, "sanity: the handler is attached"); @@ -158,6 +164,9 @@ public async Task TestUHandlerAddedThroughTheConnectionCanBeRemovedThroughTheCli client.connection.OnTransaction += handler; + // After the handler, for the reason spelled out in the test above: registration order is + // invocation order, and a witness that ran first would report a dispatch the handler had + // not seen yet. int dispatched = 0; client.connection.OnTransaction += _ => { diff --git a/Tests/Xrpl.Tests/Client/TestUStaleSessionFrames.cs b/Tests/Xrpl.Tests/Client/TestUStaleSessionFrames.cs index 3e294e08..110e5ac2 100644 --- a/Tests/Xrpl.Tests/Client/TestUStaleSessionFrames.cs +++ b/Tests/Xrpl.Tests/Client/TestUStaleSessionFrames.cs @@ -302,6 +302,11 @@ public async Task TestUFrameRefusedByACompletedChannelTakesTheFallbackPath() long? sessionId = client.connection.ActiveSessionId; Assert.IsNotNull(sessionId, "a connected client must have a session"); + // A baseline, not an absolute: the counter runs for the life of the connection, and + // the startup window this file documents elsewhere can have sent frames round the + // queue before the test got here. + long fallbackBefore = client.connection.FallbackDispatchedStreamMessages; + client.connection.CompleteStreamChannelWriterForTests(); await client.connection.IOnMessageFastPath( @@ -314,7 +319,7 @@ await client.connection.IOnMessageFastPath( Assert.AreEqual(0L, client.connection.StaleSessionFramesDropped, "the session was live throughout; nothing here is stale"); - Assert.AreEqual(1L, client.connection.FallbackDispatchedStreamMessages, + Assert.AreEqual(1L, client.connection.FallbackDispatchedStreamMessages - fallbackBefore, "a frame that went round the queue must say so - that is what the counter is for"); } finally