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