Очередь потока существует до того, как потребитель подпишется: пинг-таймер больше не гасит процессор

CHANGES.md

@@ -22,6 +22,7 @@ This release makes the SDK stop misrepresenting what a node sent. It is a delibe
 | Dropped stream events | invisible | `client.DroppedStreamMessages` counts them; `StreamMessageQueueCapacity` sizes the queue |
 | Frames from a stale or retiring session | delivered as current | dropped, and counted by `Connection.StaleSessionFramesDropped` |
 | Frames dispatched outside the queue | invisible | counted by `Connection.FallbackDispatchedStreamMessages` |
+| Frames answered while `OnConnected` runs | dispatched outside the queue, concurrently | queued like every other frame |
 | Stream event type | `LedgerStream.Type` was a public field | inherited `BaseStream.Type` property — source-compatible, but an assembly built against the old package needs a rebuild |
 | Stream event type value | `ResponseStreamType Type` | `ResponseStreamType? Type` — an event that carried no `type` no longer reports `UNKNOWN` as though the node had said so |
 
@@ -41,6 +42,12 @@ The raw-response work, all five levels landing together. The problem: a consumer
 
 The entries below are grouped by what changed, not by the order the levels were built in. The short version of what will not compile is in the table above.
 
+* **The stream queue now exists before consumer code can subscribe** (#113). `StartMessageProcessor` ran at the very end of `OnceOpen`, after `ResolveAllAwaiting()` and after the `OnConnected` callback. Subscribing from that callback is the ordinary pattern - it is what the wallet consuming this SDK does - and the node can answer the subscription before the handler returns, so the first frames found no channel and took the fallback: outside `StreamMessageQueueCapacity`, uncounted by `DroppedStreamMessages`, dispatched concurrently rather than one at a time. The events most likely to arrive out of order were precisely the first ones after connecting.
+  * **the blocker was an unrelated coupling, and that is the real fix.** `StartPingTimer` begins with `StopPingTimerSync`, which also called `StopMessageProcessor`, so simply starting the processor earlier had it torn down again moments later - measurably: `TestUDroppedStreamMessagesCountsWhatTheConsumerNeverSaw` went from 10 evictions to 0. Stopping a ping timer has no business stopping the message queue; the two lifecycles are now separate
+  * the processor is stopped explicitly where a connection genuinely ends - `Disconnect`, `DisconnectAndWaitAsync`, `OnceClose`, `OnConnectHandlerFailedAsync`, `ChangeServer` and `RetireCurrentSessionAndReconnectAsync` - rather than by side effect of a timer
+  * the fallback path stays, and that was measured rather than assumed: declaring it unreachable fails 6 tests. It still serves `OnMessage` on a client that never connected, anything arriving after the processor is stopped, and a write the channel refuses because its writer is already completed
+  * `FallbackDispatchedStreamMessages`, added alongside, is what makes this checkable from outside. It counts cumulatively and the fallback stays legitimate outside a connection, so the thing to watch is not an absolute zero but a zero *delta*: the counter should not move while a connection is being established
+
 * **The model stopped inventing values it was never sent, and stopped losing ones it does not know** (**breaking**) — the second half of the raw-response work. Re-serializing a typed response used to differ from what arrived in both directions. Measured on live mainnet responses at `api_version = 2`: a ten-entry `account_tx` gained **156 fabricated members** and dropped 28. After this level the same capture gains **4** and drops 15, and every fabricated member left is the `Amount`/`DeliverMax` rename, which is the next level.
   * **Scope came from the protocol, not from a hand count.** The repository already vendors rippled's `ledger_entries.macro` and parses which fields are Required, Optional or Default. A new conformance test, `TestUNullabilityConformance`, reads it: a field the protocol allows to be absent must map to a property that can express absence. It found **9 direct violations** — `AMM.TradingFee`, `FeeSettings.ReferenceFeeUnits/ReserveBase/ReserveIncrement`, `LedgerHashes.FirstLedgerSequence/LastLedgerSequence`, `MPTokenIssuance.AssetScale`, `PayChannel.SourceTag/DestinationTag`
   * **A second, broader rule applies to the same models.** rippled's requirement flag describes the ledger object, but these models double as the contents of `PreviousFields`, `FinalFields` and `NewFields` — and `PreviousFields` carries only the members a transaction changed, so there even a Required field can be missing. Hence every value-typed property of a ledger-entry model is now nullable. Counted as the conformance test counts them — pairs of (model, property), which is what the defect is measured in — that is **80 pairs**; as a diff it is **50 property declarations across 23 files**, the difference being `LedgerEntryType` on the shared base, which one edit covers for all 31 models. On top of that, 21 properties on transaction models and `AccountInfo.LedgerIndex`/`LedgerCurrentIndex`

Tests/Xrpl.Tests/Client/TestUStaleSessionFrames.cs

@@ -85,9 +85,10 @@ private static async Task DriveWitnessAndWait(XrplClient client, ConcurrentQueue
     /// Waits until stream frames are queued rather than taking the fallback path.
     /// </summary>
     /// <remarks>
-    /// <c>Connect()</c> returning is not enough: <c>OnceOpen</c> 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.
+    /// <c>Connect()</c> returning implies this now - <c>OnceOpen</c> starts the processor before it
+    /// resolves the waiters - so the wait is a guard rather than a necessity. It earns its place by
+    /// failing here, with a sentence saying so, if that ordering ever regresses: without it these
+    /// tests would quietly exercise the fallback path instead and still pass.
     /// </remarks>
     private static async Task WaitForMessageProcessor(XrplClient client)
     {

Tests/Xrpl.Tests/Client/TestUStreamProcessorStartsBeforeCallbacks.cs

@@ -0,0 +1,159 @@
+using Microsoft.VisualStudio.TestTools.UnitTesting;
+
+using System;
+using System.Text;
+using System.Threading;
+using System.Threading.Tasks;
+
+using Xrpl.Client;
+using Xrpl.Models.Subscriptions;
+using Xrpl.Tests;
+
+namespace XrplTests.Client;
+
+/// <summary>
+/// Frames answered while the <c>OnConnected</c> callback is still running go through the queue,
+/// like every other frame.
+/// </summary>
+/// <remarks>
+/// Subscribing from <c>OnConnected</c> is the ordinary pattern - it is what the wallet consuming
+/// this SDK does - and the node can answer that subscription before the handler returns. The
+/// message processor used to start at the very end of <c>OnceOpen</c>, after the callback, so
+/// those first frames found no channel and took the fallback: outside
+/// <c>StreamMessageQueueCapacity</c>, uncounted by <c>DroppedStreamMessages</c>, and dispatched
+/// concurrently rather than one at a time. The events most likely to arrive out of order were
+/// precisely the first ones after connecting.
+/// <para>
+/// Moving the start earlier was blocked by an unrelated coupling: <c>StartPingTimer</c> begins with
+/// <c>StopPingTimerSync</c>, which stopped the message processor as well, so an earlier start was
+/// torn down again moments later. The two lifecycles are now separate.
+/// </para>
+/// </remarks>
+[TestClass]
+public class TestUStreamProcessorStartsBeforeCallbacks
+{
+    private const string ScriptedReply =
+        "{\"id\":__ID__,\"status\":\"success\",\"type\":\"response\",\"result\":{}}";
+
+    private const string TransactionMessage = """
+    {
+      "type": "transaction",
+      "status": "closed",
+      "validated": true,
+      "engine_result": "tesSUCCESS",
+      "tx_json": { "TransactionType": "Payment", "Account": "rP9jPyP5kyvFRb6ZiRghAGw5u8SGAmU4bd", "Sequence": 4 },
+      "meta": { "AffectedNodes": [], "TransactionIndex": 0, "TransactionResult": "tesSUCCESS" }
+    }
+    """;
+
+    /// <summary>
+    /// A frame driven from inside the <c>OnConnected</c> handler reaches the queue, not the
+    /// fallback.
+    /// </summary>
+    /// <remarks>
+    /// <c>FallbackDispatchedStreamMessages</c> is the whole assertion: it counts frames dispatched
+    /// outside the queue, so before this change it would have counted this one.
+    /// </remarks>
+    [TestMethod]
+    public async Task TestUFrameAnsweredDuringOnConnectedGoesThroughTheQueue()
+    {
+        using ScriptedResponseServer server = new ScriptedResponseServer(ScriptedReply);
+        using XrplClient client = new XrplClient(server.Url);
+
+        bool channelExisted = false;
+        long fallbackDuringCallback = -1;
+        TaskCompletionSource callbackDone = new TaskCompletionSource(
+            TaskCreationOptions.RunContinuationsAsynchronously);
+
+        client.connection.OnConnected += async () =>
+        {
+            // Stands in for the node answering a subscription issued from this very handler.
+            channelExisted = client.connection.IsMessageProcessorRunning;
+            await client.connection.IOnMessageFastPath(
+                Encoding.UTF8.GetBytes(TransactionMessage),
+                client.connection.ActiveSessionId);
+            fallbackDuringCallback = client.connection.FallbackDispatchedStreamMessages;
+            callbackDone.TrySetResult();
+        };
+
+        await client.Connect();
+
+        try
+        {
+            // Connect() returning is not enough: OnceOpen resolves the waiters before invoking
+            // this callback, so without the wait the assertions below can read their defaults.
+            Task finished = await Task.WhenAny(callbackDone.Task, Task.Delay(TimeSpan.FromSeconds(5)));
+            Assert.AreSame(callbackDone.Task, finished, "the OnConnected handler never ran");
+
+            Assert.IsTrue(channelExisted,
+                "the queue did not exist while consumer code was subscribing - frames answered there take the fallback");
+            Assert.AreEqual(0L, fallbackDuringCallback,
+                "a frame answered during OnConnected went round the queue");
+        }
+        finally
+        {
+            await client.Disconnect();
+        }
+    }
+
+    /// <summary>
+    /// The queue survives the ping timer starting, which happens after the callback.
+    /// </summary>
+    /// <remarks>
+    /// This is the coupling that blocked the fix: <c>StartPingTimer</c> calls
+    /// <c>StopPingTimerSync</c>, and that used to stop the message processor too. Starting the
+    /// processor early is worth nothing if the ping timer tears it down a moment later, and the
+    /// symptom is invisible - the fallback keeps delivering frames, just without any of the queue's
+    /// guarantees.
+    /// </remarks>
+    [TestMethod]
+    public async Task TestUStartingThePingTimerLeavesTheProcessorRunning()
+    {
+        using ScriptedResponseServer server = new ScriptedResponseServer(ScriptedReply);
+        using XrplClient client = new XrplClient(
+            server.Url,
+            new XrplClient.ClientOptions { UseCustomPing = true });
+        await client.Connect();
+
+        try
+        {
+            // The ping timer starts at the very end of OnceOpen, after Connect() has returned.
+            // Asserting before it runs would pass whether or not it takes the processor down -
+            // the test has to wait for the thing that used to break it.
+            DateTime deadline = DateTime.UtcNow.AddSeconds(5);
+            while (!client.connection.IsPingTimerRunning && DateTime.UtcNow < deadline)
+            {
+                await Task.Delay(10);
+            }
+
+            Assert.IsTrue(client.connection.IsPingTimerRunning, "the ping timer never started");
+
+            Assert.IsTrue(client.connection.IsMessageProcessorRunning,
+                "the ping timer took the message processor down with it");
+
+            long fallbackBefore = client.connection.FallbackDispatchedStreamMessages;
+
+            TaskCompletionSource<TransactionStream> received = new TaskCompletionSource<TransactionStream>(
+                TaskCreationOptions.RunContinuationsAsynchronously);
+            client.OnTransaction += r =>
+            {
+                received.TrySetResult(r);
+                return Task.CompletedTask;
+            };
+
+            await client.connection.IOnMessageFastPath(
+                Encoding.UTF8.GetBytes(TransactionMessage),
+                client.connection.ActiveSessionId);
+
+            Task completed = await Task.WhenAny(received.Task, Task.Delay(TimeSpan.FromSeconds(5)));
+            Assert.AreSame(received.Task, completed, "the frame never reached the handler");
+
+            Assert.AreEqual(0L, client.connection.FallbackDispatchedStreamMessages - fallbackBefore,
+                "the frame took the fallback, so the processor was not running after all");
+        }
+        finally
+        {
+            await client.Disconnect();
+        }
+    }
+}