Один путь потока для всех платформ: браузер перестаёт обходить очередь

CHANGES.md

@@ -20,6 +20,8 @@ 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 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 |
 
@@ -203,8 +205,20 @@ 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
-  * 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
+  * **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
+  * 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
+  * **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, 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
 

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
     }
     """;
 
+    /// <summary>
+    /// Waits until the counter reaches <paramref name="expected"/>, or fails.
+    /// </summary>
+    /// <remarks>
+    /// <c>OnMessage</c> 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.
+    /// </remarks>
+    private static void WaitForCount(Func<int> 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);
+    }
+
     /// <summary>
     /// 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.
@@ -91,11 +113,29 @@ public async Task TestUHandlerAddedThroughTheClientCanBeRemovedThroughTheConnect
         // 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.
         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 += _ =>
+        {
+            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");
 
         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 +163,24 @@ 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 += _ =>
+        {
+            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");
     }