diff --git a/CHANGES.md b/CHANGES.md
index 2be04e8a..74392012 100644
--- a/CHANGES.md
+++ b/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
diff --git a/Tests/Xrpl.Tests/Client/TestUClientStreamEvents.cs b/Tests/Xrpl.Tests/Client/TestUClientStreamEvents.cs
index 91f52501..302ce755 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.
@@ -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");
}
diff --git a/Tests/Xrpl.Tests/Client/TestUStaleSessionFrames.cs b/Tests/Xrpl.Tests/Client/TestUStaleSessionFrames.cs
new file mode 100644
index 00000000..110e5ac2
--- /dev/null
+++ b/Tests/Xrpl.Tests/Client/TestUStaleSessionFrames.cs
@@ -0,0 +1,477 @@
+using Microsoft.VisualStudio.TestTools.UnitTesting;
+
+using System;
+using System.Collections.Concurrent;
+using System.Linq;
+using System.Text;
+using System.Threading;
+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" }
+ }
+ """;
+
+ ///
+ /// 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.
+ ///
+ ///
+ /// 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.
+ ///
+ [TestMethod]
+ public async Task TestUFrameFromARetiredSessionNeverReachesHandlers()
+ {
+ using ScriptedResponseServer server = new ScriptedResponseServer(ScriptedReply);
+ using XrplClient client = new XrplClient(server.Url);
+ await client.Connect();
+ await WaitForMessageProcessor(client);
+
+ ConcurrentQueue seen = new ConcurrentQueue();
+ client.OnTransaction += r =>
+ {
+ seen.Enqueue(r.Transaction.Sequence ?? 0);
+ 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");
+
+ await DriveWitnessAndWait(client, seen);
+ Assert.IsFalse(seen.Contains(DroppedSequence),
+ "the handler saw a frame belonging to a connection that is being retired");
+ }
+ finally
+ {
+ await client.Disconnect();
+ }
+ }
+
+ ///
+ /// 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();
+ await WaitForMessageProcessor(client);
+
+ ConcurrentQueue seen = new ConcurrentQueue();
+ client.OnTransaction += r =>
+ {
+ seen.Enqueue(r.Transaction.Sequence ?? 0);
+ return Task.CompletedTask;
+ };
+
+ try
+ {
+ // 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");
+
+ 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");
+
+ await DriveWitnessAndWait(client, seen);
+ Assert.IsFalse(seen.Contains(DroppedSequence),
+ "the handler saw a frame from the session being retired");
+ }
+ finally
+ {
+ await client.Disconnect();
+ }
+ }
+
+ ///
+ /// 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();
+ await WaitForMessageProcessor(client);
+
+ 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();
+ }
+ }
+
+ ///
+ /// 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");
+
+ // 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(
+ 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");
+
+ 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
+ {
+ await client.Disconnect();
+ }
+ }
+
+ ///
+ /// 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
+ /// 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();
+ }
+ }
+
+ ///
+ /// A frame naming the live session is delivered.
+ ///
+ ///
+ /// 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 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);
+ 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/Tests/Xrpl.Tests/Sugar/TestUAutofillFees.cs b/Tests/Xrpl.Tests/Sugar/TestUAutofillFees.cs
index c861f4cd..942dd377 100644
--- a/Tests/Xrpl.Tests/Sugar/TestUAutofillFees.cs
+++ b/Tests/Xrpl.Tests/Sugar/TestUAutofillFees.cs
@@ -643,6 +643,12 @@ public FeeTestClient(string feeXrp, uint reserveInc, string maxFeeXRP = "5")
public Connection connection => null!;
public long DroppedStreamMessages => 0;
+ 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 ce55ec34..d0a03128 100644
--- a/Xrpl/Client/IXrplClient.cs
+++ b/Xrpl/Client/IXrplClient.cs
@@ -67,11 +67,49 @@ 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.
///
- /// Stays at zero under WebAssembly, where frames bypass the queue entirely - see
- /// .
+ /// 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 { get; }
+ long DroppedStreamMessages => connection.DroppedStreamMessages;
+
+ ///
+ /// 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.
+ ///
+ /// 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;
+
+ ///
+ /// 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.
+ ///
+ /// 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;
/// Node error reported over the socket.
event OnError OnError;
@@ -563,6 +601,12 @@ public class ClientOptions : ConnectionOptions
///
public long DroppedStreamMessages => connection.DroppedStreamMessages;
+ ///
+ 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 eb6d24e6..4902fee6 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;
}
@@ -419,13 +415,127 @@ 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.
- private Channel? _streamMessageChannel = null;
+ // The frame travels wrapped in SessionFrame, which names the session that produced it.
+ private Channel? _streamMessageChannel = null;
private long _droppedStreamMessages;
+ 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.
+ ///
+ ///
+ /// 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);
+
+ ///
+ /// 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;
+ }
+ }
+ }
+
+ ///
+ /// 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
+ /// .
+ ///
+ ///
+ /// Volatile.Read rather than a plain read or _messageProcessorLock:
+ /// the field is written under that lock, and holding it here would mean waiting out
+ /// StopMessageProcessorInternal, which blocks up to two seconds on the reader task.
+ ///
+ internal bool IsMessageProcessorRunning => Volatile.Read(ref _streamMessageChannel) != null;
+
+ ///
+ /// 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
+ /// installation of the replacement session.
+ ///
+ ///
+ /// 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.
+ ///
+ ///
+ ///
+ /// 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.
///
@@ -440,14 +550,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;
@@ -1118,8 +1220,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)
{
@@ -1913,8 +2018,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();
}
@@ -2965,7 +3074,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,
@@ -2986,18 +3095,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);
}
}
}
@@ -3246,12 +3355,12 @@ private async Task ProcessStreamMessageAsync(byte[] frame)
///
private Task IOnMessageFastPath(string message)
{
- return IOnMessageFastPath(message, null);
+ return IOnMessageFastPath(message, null, sessionId: null);
}
///
/// 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
@@ -3262,7 +3371,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);
}
///
@@ -3284,7 +3401,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;
@@ -3365,7 +3486,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;
}
@@ -3400,52 +3521,180 @@ 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);
}
}
///
- /// 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).
+ /// 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);
+
+ ///
+ /// Whether a frame carrying this session id belongs to the connection as it stands right now.
///
- private void EnqueueStreamMessage(byte[] frame)
+ ///
+ /// 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 (OperatingSystem.IsBrowser())
+ if (sessionId is null)
{
- _ = ProcessStreamMessageFireAndForgetAsync(frame);
+ return true;
}
- else
+
+ lock (_sessionLock)
{
- 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.
- channel.Writer.TryWrite(frame);
- }
- else
- {
- _ = ProcessStreamMessageFireAndForgetAsync(frame);
- }
+ return _activeSession != null &&
+ _activeSession.SessionId == sessionId &&
+ !_activeSession.IsRetiring;
+ }
+ }
+
+ ///
+ /// 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
+ // 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.
+ //
+ // 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);
+
+ // 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)
+ {
+ return;
+ }
+
+ Interlocked.Increment(ref _fallbackDispatchedStreamMessages);
+ _ = 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.
+ /// Processes a stream frame outside the queue, on its own task.
///
- private async Task ProcessStreamMessageFireAndForgetAsync(byte[] frame)
+ ///
+ /// 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.
+ ///
+ /// 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 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);
}
}