From f3bd0ef19ed9a8d07e28e2a1b612f735593dcb59 Mon Sep 17 00:00:00 2001 From: Aleksandr <44946855+Platonenkov@users.noreply.github.com> Date: Thu, 13 Aug 2026 17:48:37 +0000 Subject: [PATCH 1/3] =?UTF-8?q?fix(json):=20=D0=B1=D0=B5=D1=81=D0=BA=D0=BE?= =?UTF-8?q?=D0=BD=D0=B5=D1=87=D0=BD=D0=B0=D1=8F=20=D1=80=D0=B5=D0=BA=D1=83?= =?UTF-8?q?=D1=80=D1=81=D0=B8=D1=8F=20=D0=B2=20LONFTokenConverter.Write=20?= =?UTF-8?q?(#87)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Метаданные любой NFT-транзакции с NFTokenPage в AffectedNodes нельзя было сериализовать: JsonSerializer.Serialize(tx.Meta) падал с "A possible object cycle was detected". Конвертер разрывал рекурсию общим для остальных полиморфных конвертеров приёмом — снять себя из options.Converters через JsonSerializerOptionsCache.WithoutConverter и войти в сериализатор заново. Приём работает только для конвертера, зарегистрированного в списке. LONFTokenConverter объявлен атрибутом [JsonConverter] на самом типе NFToken, а атрибут на типе имеет приоритет над списком, поэтому System.Text.Json возвращал значение обратно в Write независимо от содержимого списка — до упора в MaxDepth. У NFToken два поля, поэтому Write пишет их сам, не делегируя обратно. Форма JSON не меняется — {"NFToken":{"NFTokenID":"…","URI":"…"}}, — а заявленное в XML-доке поведение с null сохранено через options.DefaultIgnoreCondition вместо жёсткой политики. Остальные шесть конвертеров, вызывающих WithoutConverter, проверены по двум условиям сразу (объявлен атрибутом на типе И пересериализует тот же объявленный тип) — ни один их не выполняет; TransactionResponseConverter уже обезврежен сентинелом TransactionResponseUnknown. Ничего больше не менялось. TestULONFTokenConverter покрывал только Read — теперь закреплены форма записи, round-trip с URI и без, null под XrplJsonOptions.Default и под обычными options, многотокенная NFTokenPage и, собственно регрессия, сериализация Meta с NFTokenPage в CreatedNode.NewFields, ModifiedNode.FinalFields/PreviousFields и DeletedNode.FinalFields. --- CHANGES.md | 8 + .../Converters/LONFTokenConverterTests.cs | 186 +++++++++++++++++- .../Json/Converters/LONFTokenConverter.cs | 32 ++- Xrpl/Xrpl.csproj | 2 +- 4 files changed, 220 insertions(+), 8 deletions(-) diff --git a/CHANGES.md b/CHANGES.md index 43133aa8..7ed00e5f 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -1,5 +1,13 @@ # Changes +## 10.11.1.0 08/13/2026 + +* **Fix infinite recursion in `LONFTokenConverter.Write` — the metadata of an NFT transaction could not be serialized at all** — regression introduced in 10.3.0.0 with the `Newtonsoft.Json` → `System.Text.Json` migration; affects every release from 10.3.0.0 on. `JsonSerializer.Serialize(tx.Meta)` threw `JsonException: A possible object cycle was detected` for any transaction whose `AffectedNodes` contain an `NFTokenPage`, which is every `NFTokenMint`, `NFTokenBurn`, `NFTokenAcceptOffer` and `NFTokenModify` that touched a page. Verified against mainnet on all six NFT transaction types — the four above failed, `NFTokenCreateOffer` and `NFTokenCancelOffer` (no page in their metadata) went through: + * The converter broke its own recursion the way the other polymorphic converters do — strip itself from `options.Converters` via `JsonSerializerOptionsCache.WithoutConverter` and re-enter the serializer. That works only for a converter that is *registered in the list*. `LONFTokenConverter` is declared as a `[JsonConverter]` **attribute on the `NFToken` type itself** (`LONFTokenPage.cs`), and a converter attached to a type outranks the options list, so System.Text.Json handed the value straight back to `Write` no matter what the list looked like. The frame repeated until the writer hit `MaxDepth`. Raising `MaxDepth` is not a workaround: at 64 and 128 it is a catchable `JsonException`, at 256 the stack overflows and the process dies + * `NFToken` has two fields, so `Write` now emits them directly instead of delegating. The wire shape is unchanged — `{"NFToken":{"NFTokenID":"…","URI":"…"}}`, the envelope `Read` already looks for — and the documented null behaviour is preserved by honouring `options.DefaultIgnoreCondition` rather than hard-coding one: `XrplJsonOptions.Default` (`WhenWritingNull`) omits a null `URI`, plain options keep it as `null` + * The other six converters that call `WithoutConverter` were audited against the same two conditions — declared as a type-level attribute **and** re-serializing that same declared type. None hit both. `LOConverter` is registered in the options list (its one attribute use is property-level) and writes the concrete runtime type; `GenericStringConverter`, `MetaBinaryConverter`, `LedgerBinaryConverter` and `TransactionRequestConverter` are only ever attached to properties; the three node converters are type-level but serialize a *different* class (`value.NewFields.GetType()`). `TransactionResponseConverter` is the one other type-level case, and the same trap was already defused there by the `TransactionResponseUnknown` sentinel, so that no value ever carries the annotated type at runtime. Nothing else was changed + * `TestULONFTokenConverter` had `Read` coverage only, which is how the bug survived. It now pins the written shape, both round trips (`URI` set and null), null handling under `XrplJsonOptions.Default` and under plain options, a multi-token `NFTokenPage`, and — the regression test proper — serializing a `Meta` carrying an `NFTokenPage` in `CreatedNode.NewFields`, `ModifiedNode.FinalFields`, `ModifiedNode.PreviousFields` and `DeletedNode.FinalFields`, since the page can arrive in any of them. All offline, on prepared JSON + ## 10.11.0.0 08/04/2026 * **MPT path steps (`0x40`)** — `PathSet` only knew the three classic hop-type bits (`0x01` account, `0x10` currency, `0x20` issuer). rippled added `STPathElement::TypeMpt = 0x40` in **3.2.0**, so a hop can now carry a 24-byte `MPTokenIssuanceID` instead of a currency. The gap was silent in both directions: `FromParser` matched none of its masks on a `0x40` byte, produced an empty hop and left the 24 MPTID bytes unread — every following byte was then parsed at the wrong offset — while `SynthesizeType` had no way to emit the bit at all. Now handled end to end: diff --git a/Tests/Xrpl.Tests/Client/Json/Converters/LONFTokenConverterTests.cs b/Tests/Xrpl.Tests/Client/Json/Converters/LONFTokenConverterTests.cs index b99a8864..5be618e9 100644 --- a/Tests/Xrpl.Tests/Client/Json/Converters/LONFTokenConverterTests.cs +++ b/Tests/Xrpl.Tests/Client/Json/Converters/LONFTokenConverterTests.cs @@ -1,16 +1,21 @@ using Microsoft.VisualStudio.TestTools.UnitTesting; +using System.Collections.Generic; using System.Text.Json; using Xrpl.Client.Json; -using Xrpl.Client.Json.Converters; using Xrpl.Models.Methods; +using Xrpl.Models.Transactions; namespace XrplTests.Client.Json.Converters; [TestClass] public class TestULONFTokenConverter { + private const string NFTokenID = "000800006203F49C21D5D6E022CB16DE3538F248662FC73C29ABA6A90000000D"; + private const string OtherNFTokenID = "000800006203F49C21D5D6E022CB16DE3538F248662FC73C29ABA6A90000000E"; + private const string URI = "68747470733A2F2F6578616D706C652E636F6D"; + [TestMethod] public void Read_WrappedNFToken_UnwrapsCorrectly() { @@ -22,8 +27,8 @@ public void Read_WrappedNFToken_UnwrapsCorrectly() }"; NFToken result = JsonSerializer.Deserialize(json, XrplJsonOptions.Default); Assert.IsNotNull(result); - Assert.AreEqual("000800006203F49C21D5D6E022CB16DE3538F248662FC73C29ABA6A90000000D", result.NFTokenID); - Assert.AreEqual("68747470733A2F2F6578616D706C652E636F6D", result.URI); + Assert.AreEqual(NFTokenID, result.NFTokenID); + Assert.AreEqual(URI, result.URI); } [TestMethod] @@ -36,7 +41,180 @@ public void Read_MissingUri_NftIdOnly() }"; NFToken result = JsonSerializer.Deserialize(json, XrplJsonOptions.Default); Assert.IsNotNull(result); - Assert.AreEqual("000800006203F49C21D5D6E022CB16DE3538F248662FC73C29ABA6A90000000D", result.NFTokenID); + Assert.AreEqual(NFTokenID, result.NFTokenID); + Assert.IsNull(result.URI); + } + + /// + /// The converter is declared as an attribute on itself, which outranks + /// options.Converters — re-entering the serializer with the converter stripped from that list used to + /// call Write again, recursively, until System.Text.Json aborted at MaxDepth. + /// + [TestMethod] + public void Write_SingleNFToken_WritesWrappedShape() + { + NFToken token = new NFToken { NFTokenID = NFTokenID, URI = URI }; + + string json = JsonSerializer.Serialize(token, XrplJsonOptions.Default); + + Assert.AreEqual( + "{\"NFToken\":{\"NFTokenID\":\"" + NFTokenID + "\",\"URI\":\"" + URI + "\"}}", + json); + } + + [TestMethod] + public void Write_NullUri_OmitsUri() + { + NFToken token = new NFToken { NFTokenID = NFTokenID, URI = null }; + + string json = JsonSerializer.Serialize(token, XrplJsonOptions.Default); + + Assert.AreEqual("{\"NFToken\":{\"NFTokenID\":\"" + NFTokenID + "\"}}", json); + } + + [TestMethod] + public void Write_NullToken_WritesJsonNull() + { + Assert.AreEqual("null", JsonSerializer.Serialize(null, XrplJsonOptions.Default)); + } + + /// + /// The converter honours the ignore condition of the options it is handed instead of hard-coding one: + /// plain options keep nulls, drops them. + /// + [TestMethod] + public void Write_NullUri_PlainOptions_KeepsNull() + { + NFToken token = new NFToken { NFTokenID = NFTokenID, URI = null }; + + string json = JsonSerializer.Serialize(token, new JsonSerializerOptions()); + + Assert.AreEqual("{\"NFToken\":{\"NFTokenID\":\"" + NFTokenID + "\",\"URI\":null}}", json); + } + + [TestMethod] + public void RoundTrip_FullNFToken_Matches() + { + NFToken source = new NFToken { NFTokenID = NFTokenID, URI = URI }; + + NFToken result = JsonSerializer.Deserialize( + JsonSerializer.Serialize(source, XrplJsonOptions.Default), XrplJsonOptions.Default); + + Assert.IsNotNull(result); + Assert.AreEqual(source.NFTokenID, result.NFTokenID); + Assert.AreEqual(source.URI, result.URI); + } + + [TestMethod] + public void RoundTrip_NullUri_Matches() + { + NFToken source = new NFToken { NFTokenID = NFTokenID, URI = null }; + + NFToken result = JsonSerializer.Deserialize( + JsonSerializer.Serialize(source, XrplJsonOptions.Default), XrplJsonOptions.Default); + + Assert.IsNotNull(result); + Assert.AreEqual(source.NFTokenID, result.NFTokenID); Assert.IsNull(result.URI); } + + [TestMethod] + public void Write_NFTokenPage_SerializesEveryToken() + { + LONFTokenPage page = new LONFTokenPage + { + NFTokens = new List + { + new NFToken { NFTokenID = NFTokenID, URI = URI }, + new NFToken { NFTokenID = OtherNFTokenID }, + } + }; + + string json = JsonSerializer.Serialize(page, XrplJsonOptions.Default); + + StringAssert.Contains(json, "\"NFToken\":{\"NFTokenID\":\"" + NFTokenID + "\",\"URI\":\"" + URI + "\"}"); + StringAssert.Contains(json, "\"NFToken\":{\"NFTokenID\":\"" + OtherNFTokenID + "\"}"); + } + + /// + /// Regression: an NFTokenPage reaches through any of the three affected-node kinds, + /// so serializing the metadata of an NFToken transaction hit the recursion in every one of them. + /// + [TestMethod] + public void Write_MetaWithNFTokenPageInCreatedNode_Completes() + { + AssertMetaRoundTrips(BuildMeta(@"{ + ""CreatedNode"": { + ""LedgerEntryType"": ""NFTokenPage"", + ""LedgerIndex"": ""0FD0A2E7D0B4E7D77E5A6F9DE0D5E4A00000000000000000000000000000FFFF"", + ""NewFields"": { ""NFTokens"": [ NFTOKEN ] } + } + }")); + } + + [TestMethod] + public void Write_MetaWithNFTokenPageInModifiedNodeFinalFields_Completes() + { + AssertMetaRoundTrips(BuildMeta(@"{ + ""ModifiedNode"": { + ""LedgerEntryType"": ""NFTokenPage"", + ""LedgerIndex"": ""0FD0A2E7D0B4E7D77E5A6F9DE0D5E4A00000000000000000000000000000FFFF"", + ""FinalFields"": { ""Flags"": 0, ""NFTokens"": [ NFTOKEN ] }, + ""PreviousTxnID"": ""03F847F7728739230C2C783FE1F0D56BCFE379FEFA521053FCCFBA1F9D697255"", + ""PreviousTxnLgrSeq"": 75443929 + } + }")); + } + + [TestMethod] + public void Write_MetaWithNFTokenPageInModifiedNodePreviousFields_Completes() + { + AssertMetaRoundTrips(BuildMeta(@"{ + ""ModifiedNode"": { + ""LedgerEntryType"": ""NFTokenPage"", + ""LedgerIndex"": ""0FD0A2E7D0B4E7D77E5A6F9DE0D5E4A00000000000000000000000000000FFFF"", + ""FinalFields"": { ""Flags"": 0 }, + ""PreviousFields"": { ""NFTokens"": [ NFTOKEN ] }, + ""PreviousTxnID"": ""03F847F7728739230C2C783FE1F0D56BCFE379FEFA521053FCCFBA1F9D697255"", + ""PreviousTxnLgrSeq"": 75443929 + } + }")); + } + + [TestMethod] + public void Write_MetaWithNFTokenPageInDeletedNodeFinalFields_Completes() + { + AssertMetaRoundTrips(BuildMeta(@"{ + ""DeletedNode"": { + ""LedgerEntryType"": ""NFTokenPage"", + ""LedgerIndex"": ""0FD0A2E7D0B4E7D77E5A6F9DE0D5E4A00000000000000000000000000000FFFF"", + ""FinalFields"": { ""Flags"": 0, ""NFTokens"": [ NFTOKEN ] } + } + }")); + } + + private static Meta BuildMeta(string affectedNodeJson) + { + string nfToken = "{ \"NFToken\": { \"NFTokenID\": \"" + NFTokenID + "\", \"URI\": \"" + URI + "\" } }"; + string json = @"{ + ""TransactionIndex"": 0, + ""TransactionResult"": ""tesSUCCESS"", + ""AffectedNodes"": [ " + affectedNodeJson.Replace("NFTOKEN", nfToken) + @" ] + }"; + + Meta meta = JsonSerializer.Deserialize(json, XrplJsonOptions.Default); + Assert.IsNotNull(meta); + return meta; + } + + private static void AssertMetaRoundTrips(Meta meta) + { + string json = JsonSerializer.Serialize(meta, XrplJsonOptions.Default); + + StringAssert.Contains(json, "\"NFToken\":{\"NFTokenID\":\"" + NFTokenID + "\",\"URI\":\"" + URI + "\"}"); + + Meta reread = JsonSerializer.Deserialize(json, XrplJsonOptions.Default); + Assert.IsNotNull(reread); + Assert.AreEqual(1, reread.AffectedNodes.Count); + } } diff --git a/Xrpl/Client/Json/Converters/LONFTokenConverter.cs b/Xrpl/Client/Json/Converters/LONFTokenConverter.cs index e19f1b37..9427d44f 100644 --- a/Xrpl/Client/Json/Converters/LONFTokenConverter.cs +++ b/Xrpl/Client/Json/Converters/LONFTokenConverter.cs @@ -16,6 +16,13 @@ public class LONFTokenConverter : JsonConverter /// Writes an to JSON, wrapping it in an NFToken property. /// Null fields are ignored based on the serializer settings. /// + /// + /// The fields are written by hand rather than by re-entering the serializer. This converter is declared + /// as a on itself, and a converter attached + /// to a type outranks : dropping this converter from that + /// list — the usual way out of a recursive Write — does not stop System.Text.Json from picking it up + /// again for the same type, so Write called itself until the writer hit MaxDepth. + /// public override void Write(Utf8JsonWriter writer, NFToken value, JsonSerializerOptions options) { if (value == null) @@ -27,13 +34,32 @@ public override void Write(Utf8JsonWriter writer, NFToken value, JsonSerializerO writer.WriteStartObject(); writer.WritePropertyName("NFToken"); - // Remove this converter to avoid infinite recursion - JsonSerializerOptions innerOptions = JsonSerializerOptionsCache.WithoutConverter(options); - JsonSerializer.Serialize(writer, value, innerOptions); + writer.WriteStartObject(); + WriteField(writer, "NFTokenID", value.NFTokenID, options); + WriteField(writer, "URI", value.URI, options); + writer.WriteEndObject(); writer.WriteEndObject(); } + /// + /// Writes a single string field, honouring the ignore condition of the options in effect so that the + /// output matches what the reflection-based serializer would have produced for the same property. + /// + private static void WriteField(Utf8JsonWriter writer, string propertyName, string value, JsonSerializerOptions options) + { + if (value != null) + { + writer.WriteString(propertyName, value); + return; + } + + if (options.DefaultIgnoreCondition is JsonIgnoreCondition.WhenWritingNull or JsonIgnoreCondition.WhenWritingDefault) + return; + + writer.WriteNull(propertyName); + } + /// read from json object /// json reader diff --git a/Xrpl/Xrpl.csproj b/Xrpl/Xrpl.csproj index 8adc651e..5cc56773 100644 --- a/Xrpl/Xrpl.csproj +++ b/Xrpl/Xrpl.csproj @@ -14,7 +14,7 @@ Apache-2.0 https://github.com/StaticBit-io/XrplCSharp XrplCSharp - 10.11.0.0 + 10.11.1.0 From a219a9ebb2b405e6df5a2ce2fcc601ef0c0080dc Mon Sep 17 00:00:00 2001 From: Aleksandr <44946855+Platonenkov@users.noreply.github.com> Date: Thu, 13 Aug 2026 18:38:37 +0000 Subject: [PATCH 2/3] =?UTF-8?q?perf(client):=20=D0=BB=D0=B8=D0=BD=D0=B5?= =?UTF-8?q?=D0=B9=D0=BD=D0=B0=D1=8F=20=D1=81=D0=B1=D0=BE=D1=80=D0=BA=D0=B0?= =?UTF-8?q?=20WebSocket-=D1=81=D0=BE=D0=BE=D0=B1=D1=89=D0=B5=D0=BD=D0=B8?= =?UTF-8?q?=D0=B9=20=D0=B2=D0=BC=D0=B5=D1=81=D1=82=D0=BE=20=D0=BA=D0=B2?= =?UTF-8?q?=D0=B0=D0=B4=D1=80=D0=B0=D1=82=D0=B8=D1=87=D0=BD=D0=BE=D0=B9=20?= =?UTF-8?q?(#89)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * perf(client): линейная сборка WebSocket-сообщений вместо квадратичной ReceiveLoopAsync собирал многочанковое сообщение через byteResult.Concat(buffer.Take(result.Count)).ToArray() — на каждый чанк аллоцировался новый массив во всю накопленную длину, плюс поэлементное LINQ-перечисление вместо блочного копирования. Все промежуточные массивы крупнее 85 КБ уходили в LOH. Теперь чанки копируются Buffer.BlockCopy в scratch-буфер, который растёт до максимального сообщения соединения и дальше переиспользуется; одночанковые сообщения копируются напрямую, минуя scratch. Буфер приёма берётся из ArrayPool, а не аллоцируется на каждое соединение. Замер (300 сообщений по 2 МиБ, аллокации на сообщение): 1 чанк 3.50x payload -> 3.01x 8 чанков 6.50x -> 3.01x 32 чанка 18.52x -> 3.01x Полный стек, 3000 страниц ledger_data по 2 МиБ в 32 чанка с растущим живым heap потребителя: время 100.7 с -> 55.8 с (29.8 -> 53.8 страниц/с) аллокации 158.4 ГиБ -> 67.3 ГиБ (54.1 -> 23.0 МиБ на страницу) сборок gen2 891 -> 398 тренд последней децили к первой 1.39x -> 1.13x Заодно убрана мёртвая переменная timedOut: она объявлялась и проверялась, но никогда не присваивалась с момента появления в 3c7e38e. * fix(client): освобождать таймер таймаута запроса, а не только останавливать Resolve и Reject вызывали timer.Stop(), но не Dispose(). System.Timers.Timer наследует Component с финализатором, поэтому каждый завершённый запрос оставлял финализируемый объект; за длинный постраничный обход это тысячи недоосвобождённых таймеров, каждый из которых через замыкание Elapsed ещё и удерживает сериализованный текст своего запроса. Dispose останавливает таймер и снимает его с очереди финализации. * perf(client): убрать рефлексию с пути разбора каждого ответа Resolve, Reject и ObserveTaskException доставали TrySetResult / TrySetException / Task через GetType().GetMethod(...) + Invoke на каждый ответ. TaskInfo теперь несёт типизированные делегаты SetResult и SetException и саму CompletionTask, проставляемые при создании запроса. Свойства добавлены, а не заменены: TaskInfo — публичный тип, поэтому для экземпляров, собранных вне RequestManager, оставлен прежний путь через рефлексию. * refactor(client): точнее комментарий про пул и копирование только собранного По итогам селф-ревью: комментарий у аренды буфера говорил про «утечку», хотя речь про свежий мегабайт в LOH на каждое переподключение. При росте scratch-буфера копируется ровно собранная часть, а не весь старый массив вместе с мусором за концом. * docs(changes): запись о линейной сборке сообщений и правках RequestManager * refactor(client): убрать мёртвое поле tasks из XrplClient private readonly ConcurrentDictionary tasks нигде не присваивалось и нигде не читалось — readonly-поле без инициализации, всегда null. Остаток от реализации, где клиент сам вёл учёт запросов; сейчас этим занимается RequestManager. XrplClient не partial, обращений по имени через рефлексию нет. System.Collections.Concurrent держался только этой строкой и убран вместе с ней. * fix(client): не оставлять таймер и регистрацию отмены от уже отменённого токена Замечания CodeRabbit к PR #89. - RequestManager: уже отменённый токен выполняет колбэк Register синхронно, поэтому Reject завершал запрос раньше, чем фабрика успевала зарегистрировать таймер таймаута — и снимать было нечего. Таймер добавлялся уже после удаления промиса и оставался в timeoutsAwaitingResponse навсегда: когда он срабатывал, Reject уходил в ранний return, не дойдя до снятия. По той же причине оставалась неосвобождённой CancellationTokenRegistration — присваивание в TaskInfo происходило после того, как DeletePromise уже отработал. Обе фабрики теперь проверяют, жив ли ещё промис, и подчищают за собой; снятие таймера вынесено в DisposeTimeout и вызывается в том числе на ранних возвратах Resolve и Reject, что закрывает и узкую гонку с параллельной отменой - PagedResponseServer: offset не ограничивался длиной payload, хотя в BulkMessageServer ограничение уже стояло. При маленьком payload и большом числе фрагментов деление с округлением вверх уводило offset за конец, length уходил в минус и AsMemory бросал ArgumentOutOfRangeException мимо catch-ей AcceptAsync - TestUWebSocketMessageAssembly помечен [DoNotParallelize]: GC.GetTotalAllocatedBytes считает аллокации всего процесса, а прогон параллелит на уровне классов Тесты: TestURequestManagerCancellation — уже отменённый токен не оставляет ни таймера, ни промиса (обе фабрики), живой запрос по-прежнему взводит таймаут и снимает его при завершении. Первые два падают на коде до правки. * docs(changes): перенести записи в раздел 10.11.1.0 Раздел 10.11.0.0 уже выпущен, 10.11.1.0 открыт в dev и висит релизным PR #88 — записи изначально ушли не туда. Заодно запись про таймеры расширена случаем уже отменённого токена и добавлена запись про удалённое мёртвое поле tasks. --- CHANGES.md | 4 + .../Client/BenchmarkLedgerDataCrawl.cs | 177 +++++++++ .../Client/BenchmarkWebSocketAssembly.cs | 161 ++++++++ Tests/Xrpl.Tests/Client/BulkMessageServer.cs | 261 +++++++++++++ .../Xrpl.Tests/Client/PagedResponseServer.cs | 357 ++++++++++++++++++ .../Client/TestURequestManagerCancellation.cs | 102 +++++ .../Client/TestUWebSocketMessageAssembly.cs | 138 +++++++ Xrpl/Client/IXrplClient.cs | 3 - Xrpl/Client/RequestManager.cs | 128 ++++++- Xrpl/Client/TaskInfo.cs | 20 + Xrpl/Client/WebSocketClient.cs | 80 +++- 11 files changed, 1398 insertions(+), 33 deletions(-) create mode 100644 Tests/Xrpl.Tests/Client/BenchmarkLedgerDataCrawl.cs create mode 100644 Tests/Xrpl.Tests/Client/BenchmarkWebSocketAssembly.cs create mode 100644 Tests/Xrpl.Tests/Client/BulkMessageServer.cs create mode 100644 Tests/Xrpl.Tests/Client/PagedResponseServer.cs create mode 100644 Tests/Xrpl.Tests/Client/TestURequestManagerCancellation.cs create mode 100644 Tests/Xrpl.Tests/Client/TestUWebSocketMessageAssembly.cs diff --git a/CHANGES.md b/CHANGES.md index 7ed00e5f..e1db87e1 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -7,6 +7,10 @@ * `NFToken` has two fields, so `Write` now emits them directly instead of delegating. The wire shape is unchanged — `{"NFToken":{"NFTokenID":"…","URI":"…"}}`, the envelope `Read` already looks for — and the documented null behaviour is preserved by honouring `options.DefaultIgnoreCondition` rather than hard-coding one: `XrplJsonOptions.Default` (`WhenWritingNull`) omits a null `URI`, plain options keep it as `null` * The other six converters that call `WithoutConverter` were audited against the same two conditions — declared as a type-level attribute **and** re-serializing that same declared type. None hit both. `LOConverter` is registered in the options list (its one attribute use is property-level) and writes the concrete runtime type; `GenericStringConverter`, `MetaBinaryConverter`, `LedgerBinaryConverter` and `TransactionRequestConverter` are only ever attached to properties; the three node converters are type-level but serialize a *different* class (`value.NewFields.GetType()`). `TransactionResponseConverter` is the one other type-level case, and the same trap was already defused there by the `TransactionResponseUnknown` sentinel, so that no value ever carries the annotated type at runtime. Nothing else was changed * `TestULONFTokenConverter` had `Read` coverage only, which is how the bug survived. It now pins the written shape, both round trips (`URI` set and null), null handling under `XrplJsonOptions.Default` and under plain options, a multi-token `NFTokenPage`, and — the regression test proper — serializing a `Meta` carrying an `NFTokenPage` in `CreatedNode.NewFields`, `ModifiedNode.FinalFields`, `ModifiedNode.PreviousFields` and `DeletedNode.FinalFields`, since the page can arrive in any of them. All offline, on prepared JSON +* **WebSocket message assembly was quadratic in the number of receive chunks** — `ReceiveLoopAsync` grew a multi-chunk message with `byteResult = byteResult.Concat(buffer.Take(result.Count)).ToArray()`. Every chunk allocated a fresh array the size of everything received so far and refilled it one byte at a time through a LINQ enumerator, so a message split into *k* chunks copied roughly `k/2` times its own length; every intermediate array was well past the 85 KB threshold and therefore landed on the uncompacted large object heap. `ledger_data` at `limit=2048` is a few megabytes and arrives in dozens of chunks over a real link, which is exactly where the cost concentrates. Chunks are now `Buffer.BlockCopy`-ed into a scratch buffer that grows to the largest message on the connection and is reused from then on; a message that arrives whole in one chunk skips the scratch entirely, and the receive buffer itself is rented from `ArrayPool` rather than allocated per connection. Measured on a local fragmenting WebSocket server, 300 messages of 2 MiB, allocation per message: 3.50x payload at one chunk, 6.50x at eight, 18.52x at thirty-two — now a flat 3.01x, which is the floor (the exact-sized `byte[]` plus the UTF-16 string handed to the callback). Through the full client stack, a 3000-page `ledger_data` crawl with each page arriving in 32 chunks and a consumer retaining every object: 100.7 s → 55.8 s, 158.4 GiB → 67.3 GiB allocated, 891 → 398 gen2 collections, and the last-decile-to-first-decile page time drops from 1.39x to 1.13x. `ReceiveChunkSize` was measured at 1 MiB and 64 KiB and left at 1 MiB — now that the buffer is pooled, shrinking it changed nothing outside run-to-run noise. `TestUWebSocketMessageAssembly` pins the byte-exactness of a 96-chunk message, that a short message after a long one picks up no stale bytes from the reused buffer, and that per-message allocation stays under 12x payload at 64 chunks (34.5x before the fix). A dead `timedOut` local, declared and tested but never assigned since it appeared, is gone +* **Request timeout timers outlived their requests** — `RequestManager.Resolve`/`Reject` called `timer.Stop()`. `System.Timers.Timer` derives from `Component` and carries a finalizer, so every completed request left a finalizable object behind, each of them holding its request's serialized text alive through the `Elapsed` closure; over a long paged crawl that is thousands of them. `Dispose()` stops the timer and takes it off the finalization queue. A second, worse case sat next to it: a token that is **already cancelled** runs its `Register` callback inline, so `Reject` completed the request in the middle of the factory method — before the timeout timer existed and therefore with nothing to remove. The factory then registered the timer for a promise that was already gone, and when it fired, `Reject` took its missing-promise early return without removing it, so the entry stayed in `timeoutsAwaitingResponse` for the life of the process. The `CancellationTokenRegistration` leaked on the same path, its assignment to `TaskInfo` happening after `DeletePromise` had already run. Both factories now check whether the promise survived and clean up after themselves; timer removal moved into `DisposeTimeout`, which is also called on the early returns of `Resolve` and `Reject` and so closes the narrow race with a concurrent cancellation as well. `TestURequestManagerCancellation` pins that an already-cancelled token leaves neither timer nor promise behind in either factory, and that a live request still arms its timeout and releases it on completion +* **Reflection on the per-response path is gone** — `Resolve`, `Reject` and `ObserveTaskException` reached for `TrySetResult`, `TrySetException` and `Task` through `GetType().GetMethod(...)` + `Invoke` on every single response. `TaskInfo` now carries typed `SetResult`/`SetException` delegates and the `CompletionTask` itself, wired when the request is created. The properties were added rather than substituted: `TaskInfo` is public, so instances built outside `RequestManager` keep the old reflective path +* **Dead `tasks` field removed from `XrplClient`** — `private readonly ConcurrentDictionary tasks` was never assigned and never read, so it was permanently null; a leftover from when the client tracked pending requests itself, which `RequestManager` has done for a long time ## 10.11.0.0 08/04/2026 diff --git a/Tests/Xrpl.Tests/Client/BenchmarkLedgerDataCrawl.cs b/Tests/Xrpl.Tests/Client/BenchmarkLedgerDataCrawl.cs new file mode 100644 index 00000000..dc86527a --- /dev/null +++ b/Tests/Xrpl.Tests/Client/BenchmarkLedgerDataCrawl.cs @@ -0,0 +1,177 @@ +using Microsoft.VisualStudio.TestTools.UnitTesting; + +using System; +using System.Collections.Generic; +using System.Diagnostics; +using System.Threading.Tasks; + +using Xrpl.Client; + +namespace Xrpl.Tests.ClientLib +{ + /// + /// Manual benchmark of a long paged crawl through the full client stack (socket receive loop, + /// message routing, RequestManager, JSON round-trip). Deliberately named outside the + /// TestU/TestI filters so it never runs in CI; invoke it explicitly: + /// dotnet test --filter "FullyQualifiedName~BenchmarkLedgerDataCrawl". + /// Knobs: CRAWL_PAGES, CRAWL_PAYLOAD_BYTES, CRAWL_FRAGMENTS, CRAWL_RETAIN_PER_PAGE. + /// CRAWL_RETAIN_PER_PAGE models a consumer that keeps every crawled object alive (a full + /// ledger-state snapshot), which is what makes each forced gen2 collection progressively + /// more expensive as the crawl advances. + /// + [TestClass] + public class BenchmarkLedgerDataCrawl + { + private static int EnvInt(string name, int fallback) + { + string? raw = Environment.GetEnvironmentVariable(name); + return int.TryParse(raw, out int value) && value >= 0 ? value : fallback; + } + + /// Stand-in for one crawled ledger object the consumer keeps in its snapshot. + private sealed class RetainedEntry + { + public RetainedEntry(string index) + { + Index = index; + } + + public string Index { get; } + } + + [TestMethod] + public async Task BenchmarkSequentialPaging() + { + int pages = EnvInt("CRAWL_PAGES", 2000); + int payloadBytes = EnvInt("CRAWL_PAYLOAD_BYTES", 2 * 1024 * 1024); + int fragments = EnvInt("CRAWL_FRAGMENTS", 32); + int retainPerPage = EnvInt("CRAWL_RETAIN_PER_PAGE", 0); + + List snapshot = new List(pages * retainPerPage); + + using PagedResponseServer server = new PagedResponseServer(payloadBytes, fragments); + using XrplClient client = new XrplClient(server.Url); + + await client.Connect().ConfigureAwait(false); + + // One warm-up page so JIT and pooled buffers are not charged to the measured window. + await RequestPageAsync(client).ConfigureAwait(false); + + GC.Collect(2, GCCollectionMode.Forced, blocking: true); + GC.WaitForPendingFinalizers(); + GC.Collect(2, GCCollectionMode.Forced, blocking: true); + + long allocatedBefore = GC.GetTotalAllocatedBytes(precise: true); + int gen0Before = GC.CollectionCount(0); + int gen1Before = GC.CollectionCount(1); + int gen2Before = GC.CollectionCount(2); + long heapBefore = GC.GetTotalMemory(false); + long lohBefore = LohBytes(); + + double[] pageMs = new double[pages]; + long startTicks = Stopwatch.GetTimestamp(); + + for (int i = 0; i < pages; i++) + { + long before = Stopwatch.GetTimestamp(); + await RequestPageAsync(client).ConfigureAwait(false); + + for (int entry = 0; entry < retainPerPage; entry++) + { + snapshot.Add(new RetainedEntry(((long)i * retainPerPage + entry).ToString("X16"))); + } + + pageMs[i] = (Stopwatch.GetTimestamp() - before) * 1000.0 / Stopwatch.Frequency; + } + + double totalSeconds = (Stopwatch.GetTimestamp() - startTicks) / (double)Stopwatch.Frequency; + long allocated = GC.GetTotalAllocatedBytes(precise: true) - allocatedBefore; + long heapAfter = GC.GetTotalMemory(false); + long lohAfter = LohBytes(); + + await client.Disconnect().ConfigureAwait(false); + + Console.WriteLine("=== ledger_data crawl benchmark (full client stack) ==="); + Console.WriteLine($"pages : {pages}"); + Console.WriteLine($"payload : {payloadBytes / 1024.0 / 1024.0:F2} MiB"); + Console.WriteLine($"fragments per page: {fragments}"); + Console.WriteLine($"retained objects : {snapshot.Count:N0} ({retainPerPage}/page)"); + Console.WriteLine($"total time : {totalSeconds:F2} s ({pages / totalSeconds:F2} pages/s)"); + Console.WriteLine($"first decile avg : {DecileAverage(pageMs, 0):F1} ms/page"); + Console.WriteLine($"last decile avg : {DecileAverage(pageMs, 9):F1} ms/page"); + Console.WriteLine($"trend (last/first): {DecileAverage(pageMs, 9) / DecileAverage(pageMs, 0):F2}x"); + Console.WriteLine($"p50 / p99 / max : {Percentile(pageMs, 50):F1} / {Percentile(pageMs, 99):F1} / " + + $"{Percentile(pageMs, 100):F1} ms"); + Console.WriteLine($"allocated total : {allocated / 1024.0 / 1024.0 / 1024.0:F2} GiB"); + Console.WriteLine($"allocated per page: {allocated / (double)pages / 1024.0 / 1024.0:F1} MiB " + + $"({allocated / (double)pages / payloadBytes:F1}x payload)"); + Console.WriteLine($"gen0/gen1/gen2 : {GC.CollectionCount(0) - gen0Before} / " + + $"{GC.CollectionCount(1) - gen1Before} / {GC.CollectionCount(2) - gen2Before}"); + Console.WriteLine($"managed heap : {heapBefore / 1024.0 / 1024.0:F1} -> {heapAfter / 1024.0 / 1024.0:F1} MiB"); + Console.WriteLine($"LOH size : {lohBefore / 1024.0 / 1024.0:F1} -> {lohAfter / 1024.0 / 1024.0:F1} MiB"); + Console.WriteLine("decile profile (ms/page): " + string.Join(" | ", DecileProfile(pageMs))); + } + + private static async Task RequestPageAsync(XrplClient client) + { + Dictionary request = new Dictionary + { + ["command"] = "ledger_data", + ["ledger_index"] = 96000000, + ["binary"] = true, + ["limit"] = 2048 + }; + + Dictionary response = await client.Request(request).ConfigureAwait(false); + if (response == null) + { + throw new InvalidOperationException("empty ledger_data response"); + } + } + + private static long LohBytes() + { + GCMemoryInfo info = GC.GetGCMemoryInfo(); + ReadOnlySpan generations = info.GenerationInfo; + return generations.Length > 3 ? generations[3].SizeAfterBytes : 0; + } + + private static double DecileAverage(double[] values, int decile) + { + int size = Math.Max(1, values.Length / 10); + int from = decile * size; + int to = Math.Min(values.Length, from + size); + if (from >= to) + { + return 0; + } + + double sum = 0; + for (int i = from; i < to; i++) + { + sum += values[i]; + } + + return sum / (to - from); + } + + private static string[] DecileProfile(double[] values) + { + string[] profile = new string[10]; + for (int i = 0; i < 10; i++) + { + profile[i] = DecileAverage(values, i).ToString("F1"); + } + + return profile; + } + + private static double Percentile(double[] values, int percentile) + { + double[] sorted = (double[])values.Clone(); + Array.Sort(sorted); + int index = (int)Math.Round((percentile / 100.0) * (sorted.Length - 1)); + return sorted[Math.Clamp(index, 0, sorted.Length - 1)]; + } + } +} diff --git a/Tests/Xrpl.Tests/Client/BenchmarkWebSocketAssembly.cs b/Tests/Xrpl.Tests/Client/BenchmarkWebSocketAssembly.cs new file mode 100644 index 00000000..75f6e4c7 --- /dev/null +++ b/Tests/Xrpl.Tests/Client/BenchmarkWebSocketAssembly.cs @@ -0,0 +1,161 @@ +using Microsoft.VisualStudio.TestTools.UnitTesting; + +using System; +using System.Diagnostics; +using System.Threading; +using System.Threading.Tasks; + +using Xrpl.Client; + +namespace Xrpl.Tests.ClientLib +{ + /// + /// Manual benchmark of the WebSocket multi-chunk message assembly path. Deliberately named + /// outside the TestU/TestI filters so it never runs in CI; invoke it explicitly: + /// dotnet test --filter "FullyQualifiedName~BenchmarkWebSocketAssembly". + /// Knobs: BENCH_MESSAGES, BENCH_PAYLOAD_BYTES, BENCH_FRAGMENTS. + /// + [TestClass] + public class BenchmarkWebSocketAssembly + { + private static int EnvInt(string name, int fallback) + { + string? raw = Environment.GetEnvironmentVariable(name); + return int.TryParse(raw, out int value) && value > 0 ? value : fallback; + } + + [TestMethod] + public async Task BenchmarkMultiChunkAssembly() + { + int messageCount = EnvInt("BENCH_MESSAGES", 600); + int payloadBytes = EnvInt("BENCH_PAYLOAD_BYTES", 2 * 1024 * 1024); + int fragments = EnvInt("BENCH_FRAGMENTS", 32); + + using BulkMessageServer server = new BulkMessageServer(messageCount, payloadBytes, fragments); + + long[] timestamps = new long[messageCount]; + int received = 0; + int corrupted = 0; + TaskCompletionSource allReceived = + new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + + WebSocketClient client = WebSocketClient.Create(server.Url); + client.OnMessageReceived((message, _) => + { + int index = received; + if (message.Length != payloadBytes) + { + Interlocked.Increment(ref corrupted); + } + + if (index < timestamps.Length) + { + timestamps[index] = Stopwatch.GetTimestamp(); + } + + received = index + 1; + if (received >= messageCount) + { + allReceived.TrySetResult(true); + } + + return Task.CompletedTask; + }); + + // Warm up the JIT and the socket path before the measured window. + await client.Connect().ConfigureAwait(false); + + GC.Collect(2, GCCollectionMode.Forced, blocking: true); + GC.WaitForPendingFinalizers(); + GC.Collect(2, GCCollectionMode.Forced, blocking: true); + + long allocatedBefore = GC.GetTotalAllocatedBytes(precise: true); + int gen0Before = GC.CollectionCount(0); + int gen1Before = GC.CollectionCount(1); + int gen2Before = GC.CollectionCount(2); + long lohBefore = LohBytes(); + long startTicks = Stopwatch.GetTimestamp(); + + // Releases the server; nothing has been sent before this point. + client.SendMessage("go"); + + Task completed = await Task.WhenAny( + allReceived.Task, + Task.Delay(TimeSpan.FromMinutes(30))).ConfigureAwait(false); + + long stopTicks = Stopwatch.GetTimestamp(); + long allocatedAfter = GC.GetTotalAllocatedBytes(precise: true); + long lohAfter = LohBytes(); + + client.CancelIntentionally(); + client.Dispose(); + + Assert.AreSame(allReceived.Task, completed, "benchmark did not finish within 30 minutes"); + Assert.AreEqual(0, corrupted, "at least one message was assembled with the wrong length"); + + double totalSeconds = (stopTicks - startTicks) / (double)Stopwatch.Frequency; + long allocated = allocatedAfter - allocatedBefore; + + double firstDecile = DecileAverageMs(timestamps, startTicks, 0); + double lastDecile = DecileAverageMs(timestamps, startTicks, 9); + double slowest = SlowestMs(timestamps, startTicks); + + Console.WriteLine("=== WebSocket multi-chunk assembly benchmark ==="); + Console.WriteLine($"messages : {messageCount}"); + Console.WriteLine($"payload : {payloadBytes / 1024.0 / 1024.0:F2} MiB"); + Console.WriteLine($"fragments per msg : {fragments} ({payloadBytes / fragments / 1024} KiB each)"); + Console.WriteLine($"total time : {totalSeconds:F2} s ({messageCount / totalSeconds:F2} msg/s)"); + Console.WriteLine($"first decile avg : {firstDecile:F2} ms/msg"); + Console.WriteLine($"last decile avg : {lastDecile:F2} ms/msg"); + Console.WriteLine($"slowest message : {slowest:F2} ms"); + Console.WriteLine($"trend (last/first): {lastDecile / firstDecile:F2}x"); + Console.WriteLine($"allocated total : {allocated / 1024.0 / 1024.0:F1} MiB"); + Console.WriteLine($"allocated per msg : {allocated / (double)messageCount / 1024.0 / 1024.0:F2} MiB " + + $"({allocated / (double)messageCount / payloadBytes:F2}x payload)"); + Console.WriteLine($"gen0/gen1/gen2 : {GC.CollectionCount(0) - gen0Before} / " + + $"{GC.CollectionCount(1) - gen1Before} / {GC.CollectionCount(2) - gen2Before}"); + Console.WriteLine($"LOH size : {lohBefore / 1024.0 / 1024.0:F1} MiB -> {lohAfter / 1024.0 / 1024.0:F1} MiB"); + } + + private static long LohBytes() + { + GCMemoryInfo info = GC.GetGCMemoryInfo(); + ReadOnlySpan generations = info.GenerationInfo; + return generations.Length > 3 ? generations[3].SizeAfterBytes : 0; + } + + private static double DecileAverageMs(long[] timestamps, long startTicks, int decile) + { + int size = Math.Max(1, timestamps.Length / 10); + int from = decile * size; + int to = Math.Min(timestamps.Length, from + size); + if (from >= to) + { + return 0; + } + + long previous = from == 0 ? startTicks : timestamps[from - 1]; + double totalMs = (timestamps[to - 1] - previous) * 1000.0 / Stopwatch.Frequency; + return totalMs / (to - from); + } + + private static double SlowestMs(long[] timestamps, long startTicks) + { + double slowest = 0; + long previous = startTicks; + + foreach (long timestamp in timestamps) + { + double ms = (timestamp - previous) * 1000.0 / Stopwatch.Frequency; + if (ms > slowest) + { + slowest = ms; + } + + previous = timestamp; + } + + return slowest; + } + } +} diff --git a/Tests/Xrpl.Tests/Client/BulkMessageServer.cs b/Tests/Xrpl.Tests/Client/BulkMessageServer.cs new file mode 100644 index 00000000..62445f30 --- /dev/null +++ b/Tests/Xrpl.Tests/Client/BulkMessageServer.cs @@ -0,0 +1,261 @@ +using System; +using System.Net; +using System.Net.Sockets; +using System.Text; +using System.Threading; +using System.Threading.Tasks; + +using Xrpl.Tests.MockRippled; + +namespace Xrpl.Tests +{ + /// + /// Minimal WebSocket server that completes a handshake and then pushes a fixed number of + /// large text messages, each split into a controlled number of WebSocket continuation + /// frames. Fragmenting at the protocol level (rather than relying on how the socket happens + /// to slice the stream) makes the number of client-side receive chunks per message exact and + /// reproducible, which is what the assembly path is sensitive to. + /// + internal sealed class BulkMessageServer : IDisposable + { + private readonly TcpListener _listener; + private readonly CancellationTokenSource _cts = new(); + private readonly int _messageCount; + private readonly int _fragments; + private readonly int[] _lengthCycle; + private readonly byte[] _payload; + private readonly TaskCompletionSource _finished = + new(TaskCreationOptions.RunContinuationsAsynchronously); + + /// How many messages to push once the client connects. + /// Size of the longest message payload, in bytes. + /// + /// Number of WebSocket frames each message is split into; the client sees exactly this + /// many receive chunks per message. + /// + /// + /// Optional cycle of message lengths, each a prefix of the full payload. Lets a test mix + /// long and short messages on one connection, which is what exposes stale bytes left in a + /// reused assembly buffer. Defaults to every message being the full payload. + /// + public BulkMessageServer(int messageCount, int payloadBytes, int fragments, int[]? lengthCycle = null) + { + _messageCount = messageCount; + _fragments = Math.Max(1, fragments); + _payload = BuildPayload(payloadBytes); + _lengthCycle = lengthCycle is { Length: > 0 } ? lengthCycle : new[] { _payload.Length }; + + _listener = new TcpListener(IPAddress.Loopback, 0); + _listener.Start(); + Port = ((IPEndPoint)_listener.LocalEndpoint).Port; + _ = AcceptAsync(); + } + + public int Port { get; } + + public string Url => "ws://127.0.0.1:" + Port + "/"; + + /// Payload every message carries, as the client should see it. + public string PayloadText => Encoding.UTF8.GetString(_payload); + + public int PayloadBytes => _payload.Length; + + /// Completes with the number of messages written once the server is done sending. + public Task SendCompleted => _finished.Task; + + /// + /// Builds an ASCII payload shaped like a paged rippled response, so the byte content is + /// non-uniform and any accidental truncation during assembly is visible. + /// + private static byte[] BuildPayload(int payloadBytes) + { + StringBuilder builder = new StringBuilder(payloadBytes + 64); + builder.Append("{\"id\":1,\"status\":\"success\",\"type\":\"response\",\"result\":{\"state\":["); + + int index = 0; + while (builder.Length < payloadBytes - 80) + { + if (index > 0) + { + builder.Append(','); + } + + builder.Append("{\"i\":").Append(index).Append(",\"d\":\""); + builder.Append((char)('A' + (index % 26)), 48); + builder.Append("\"}"); + index++; + } + + builder.Append("]}}"); + + while (builder.Length < payloadBytes) + { + builder.Append(' '); + } + + return Encoding.UTF8.GetBytes(builder.ToString(0, payloadBytes)); + } + + private async Task AcceptAsync() + { + try + { + using TcpClient client = await _listener.AcceptTcpClientAsync(_cts.Token).ConfigureAwait(false); + client.NoDelay = true; + NetworkStream stream = client.GetStream(); + + string request = await ReadUntilHeadersEndAsync(stream).ConfigureAwait(false); + string key = Helpers.GetHandshakeRequestKey(request); + byte[] response = Encoding.ASCII.GetBytes(Helpers.GetHandshakeResponse(Helpers.HashKey(key))); + await stream.WriteAsync(response, _cts.Token).ConfigureAwait(false); + await stream.FlushAsync(_cts.Token).ConfigureAwait(false); + + // Wait for the client's go-ahead before pushing anything, so no message can land + // before the caller has opened its measurement window. + byte[] goAhead = new byte[256]; + if (await stream.ReadAsync(goAhead, _cts.Token).ConfigureAwait(false) == 0) + { + _finished.TrySetResult(0); + return; + } + + // Drain and discard whatever the client sends afterwards (keep-alive pings, close + // frames) so its socket never blocks on a full send window. + _ = DrainAsync(stream); + + for (int i = 0; i < _messageCount; i++) + { + int messageLength = _lengthCycle[i % _lengthCycle.Length]; + int fragmentBytes = (messageLength + _fragments - 1) / _fragments; + + for (int fragment = 0; fragment < _fragments; fragment++) + { + // Ceil division can leave trailing frames past the end; they are sent empty + // so the frame count stays exactly as requested. + int offset = Math.Min(fragment * fragmentBytes, messageLength); + int length = Math.Min(fragmentBytes, messageLength - offset); + bool isFirst = fragment == 0; + bool isLast = fragment == _fragments - 1; + + await stream.WriteAsync(BuildFrameHeader(length, isFirst, isLast), _cts.Token) + .ConfigureAwait(false); + await stream.WriteAsync(_payload.AsMemory(offset, length), _cts.Token) + .ConfigureAwait(false); + await stream.FlushAsync(_cts.Token).ConfigureAwait(false); + } + } + + _finished.TrySetResult(_messageCount); + + await Task.Delay(Timeout.InfiniteTimeSpan, _cts.Token).ConfigureAwait(false); + } + catch (OperationCanceledException) + { + _finished.TrySetCanceled(); + } + catch (Exception ex) + { + _finished.TrySetException(ex); + } + } + + private async Task DrainAsync(NetworkStream stream) + { + byte[] sink = new byte[4096]; + + try + { + while (await stream.ReadAsync(sink, _cts.Token).ConfigureAwait(false) > 0) + { + } + } + catch + { + // The connection going away is the normal end of this loop. + } + } + + /// + /// Unmasked server-to-client frame header. The first frame of a message carries the text + /// opcode (0x1), every following frame carries continuation (0x0); FIN is set on the last. + /// + private static byte[] BuildFrameHeader(int payloadLength, bool isFirst, bool isLast) + { + byte first = (byte)((isLast ? 0x80 : 0x00) | (isFirst ? 0x01 : 0x00)); + + if (payloadLength <= 125) + { + return new byte[] { first, (byte)payloadLength }; + } + + if (payloadLength <= ushort.MaxValue) + { + return new byte[] + { + first, + 126, + (byte)(payloadLength >> 8), + (byte)payloadLength + }; + } + + return new byte[] + { + first, + 127, + 0, 0, 0, 0, + (byte)(payloadLength >> 24), + (byte)(payloadLength >> 16), + (byte)(payloadLength >> 8), + (byte)payloadLength + }; + } + + private async Task ReadUntilHeadersEndAsync(NetworkStream stream) + { + byte[] buffer = new byte[4096]; + StringBuilder request = new StringBuilder(); + + while (true) + { + int read = await stream.ReadAsync(buffer, _cts.Token).ConfigureAwait(false); + if (read == 0) + { + break; + } + + request.Append(Encoding.ASCII.GetString(buffer, 0, read)); + + if (request.ToString().Contains("\r\n\r\n", StringComparison.Ordinal)) + { + break; + } + } + + return request.ToString(); + } + + public void Dispose() + { + try + { + _cts.Cancel(); + } + catch + { + // best effort + } + + try + { + _listener.Stop(); + } + catch + { + // best effort + } + + _cts.Dispose(); + } + } +} diff --git a/Tests/Xrpl.Tests/Client/PagedResponseServer.cs b/Tests/Xrpl.Tests/Client/PagedResponseServer.cs new file mode 100644 index 00000000..8162114c --- /dev/null +++ b/Tests/Xrpl.Tests/Client/PagedResponseServer.cs @@ -0,0 +1,357 @@ +using System; +using System.Buffers.Binary; +using System.IO; +using System.Net; +using System.Net.Sockets; +using System.Text; +using System.Threading; +using System.Threading.Tasks; + +using Xrpl.Tests.MockRippled; + +namespace Xrpl.Tests +{ + /// + /// Minimal WebSocket server that answers every client request with a large, paged + /// rippled-shaped response echoing the request id. Each response is split into a controlled + /// number of WebSocket continuation frames, so the client assembles it from exactly that many + /// receive chunks — which is what a multi-megabyte ledger_data page looks like over a + /// real link. + /// + internal sealed class PagedResponseServer : IDisposable + { + private readonly TcpListener _listener; + private readonly CancellationTokenSource _cts = new(); + private readonly int _fragments; + private readonly string _resultBody; + private int _served; + + /// Target size of each response, in bytes. + /// Number of WebSocket frames each response is split into. + public PagedResponseServer(int approximatePayloadBytes, int fragments) + { + _fragments = Math.Max(1, fragments); + _resultBody = BuildResultBody(approximatePayloadBytes); + + _listener = new TcpListener(IPAddress.Loopback, 0); + _listener.Start(); + Port = ((IPEndPoint)_listener.LocalEndpoint).Port; + _ = AcceptAsync(); + } + + public int Port { get; } + + public string Url => "ws://127.0.0.1:" + Port + "/"; + + /// Number of requests answered so far. + public int Served => Volatile.Read(ref _served); + + /// + /// Body of the result object: a binary-form ledger_data page, i.e. a list of + /// {data, index} pairs, which is what a bulk ledger crawl actually receives. + /// + private static string BuildResultBody(int approximatePayloadBytes) + { + StringBuilder builder = new StringBuilder(approximatePayloadBytes + 256); + builder.Append("{\"ledger_hash\":\"842B57C1CC0613299A686D3E9F310EC0422C84D3911E5056389AA7E5808A93C8\","); + builder.Append("\"ledger_index\":\"96000000\",\"validated\":true,\"state\":["); + + int index = 0; + while (builder.Length < approximatePayloadBytes) + { + if (index > 0) + { + builder.Append(','); + } + + builder.Append("{\"data\":\""); + AppendHex(builder, index, 900); + builder.Append("\",\"index\":\""); + AppendHex(builder, index + 7, 64); + builder.Append("\"}"); + index++; + } + + builder.Append("],\"marker\":\""); + AppendHex(builder, index, 64); + builder.Append("\"}"); + + return builder.ToString(); + } + + private static void AppendHex(StringBuilder builder, int seed, int length) + { + const string Digits = "0123456789ABCDEF"; + int state = (seed * 1103515245) ^ 0x5F3A; + + for (int i = 0; i < length; i++) + { + state = (state * 1103515245) + 12345; + builder.Append(Digits[(state >> 16) & 0xF]); + } + } + + private async Task AcceptAsync() + { + try + { + using TcpClient client = await _listener.AcceptTcpClientAsync(_cts.Token).ConfigureAwait(false); + client.NoDelay = true; + NetworkStream stream = client.GetStream(); + + string request = await ReadUntilHeadersEndAsync(stream).ConfigureAwait(false); + string key = Helpers.GetHandshakeRequestKey(request); + byte[] response = Encoding.ASCII.GetBytes(Helpers.GetHandshakeResponse(Helpers.HashKey(key))); + await stream.WriteAsync(response, _cts.Token).ConfigureAwait(false); + await stream.FlushAsync(_cts.Token).ConfigureAwait(false); + + while (!_cts.IsCancellationRequested) + { + string? message = await ReadTextFrameAsync(stream).ConfigureAwait(false); + if (message == null) + { + return; + } + + string id = ExtractId(message); + await WriteResponseAsync(stream, id).ConfigureAwait(false); + Interlocked.Increment(ref _served); + } + } + catch (OperationCanceledException) + { + // Normal shutdown. + } + catch (Exception ex) when (ex is IOException or SocketException or ObjectDisposedException) + { + // The client going away is the normal end of this loop. + } + } + + /// Pulls the JSON string value of the request's "id" property. + private static string ExtractId(string message) + { + int keyIndex = message.IndexOf("\"id\"", StringComparison.Ordinal); + if (keyIndex < 0) + { + return "\"0\""; + } + + int colon = message.IndexOf(':', keyIndex); + if (colon < 0) + { + return "\"0\""; + } + + int start = colon + 1; + while (start < message.Length && char.IsWhiteSpace(message[start])) + { + start++; + } + + if (start < message.Length && message[start] == '"') + { + int end = message.IndexOf('"', start + 1); + return end < 0 ? "\"0\"" : message.Substring(start, end - start + 1); + } + + int stop = start; + while (stop < message.Length && message[stop] != ',' && message[stop] != '}') + { + stop++; + } + + return message.Substring(start, stop - start).Trim(); + } + + private async Task WriteResponseAsync(NetworkStream stream, string id) + { + string envelope = "{\"id\":" + id + ",\"status\":\"success\",\"type\":\"response\",\"result\":" + + _resultBody + "}"; + byte[] payload = Encoding.UTF8.GetBytes(envelope); + int fragmentBytes = (payload.Length + _fragments - 1) / _fragments; + + for (int fragment = 0; fragment < _fragments; fragment++) + { + // Ceil division can leave trailing frames past the end; they are sent empty + // so the frame count stays exactly as requested. + int offset = Math.Min(fragment * fragmentBytes, payload.Length); + int length = Math.Min(fragmentBytes, payload.Length - offset); + bool isFirst = fragment == 0; + bool isLast = fragment == _fragments - 1; + + await stream.WriteAsync(BuildFrameHeader(length, isFirst, isLast), _cts.Token) + .ConfigureAwait(false); + await stream.WriteAsync(payload.AsMemory(offset, length), _cts.Token).ConfigureAwait(false); + await stream.FlushAsync(_cts.Token).ConfigureAwait(false); + } + } + + /// + /// Unmasked server-to-client frame header. The first frame of a message carries the text + /// opcode (0x1), every following frame carries continuation (0x0); FIN is set on the last. + /// + private static byte[] BuildFrameHeader(int payloadLength, bool isFirst, bool isLast) + { + byte first = (byte)((isLast ? 0x80 : 0x00) | (isFirst ? 0x01 : 0x00)); + + if (payloadLength <= 125) + { + return new byte[] { first, (byte)payloadLength }; + } + + if (payloadLength <= ushort.MaxValue) + { + return new byte[] { first, 126, (byte)(payloadLength >> 8), (byte)payloadLength }; + } + + return new byte[] + { + first, + 127, + 0, 0, 0, 0, + (byte)(payloadLength >> 24), + (byte)(payloadLength >> 16), + (byte)(payloadLength >> 8), + (byte)payloadLength + }; + } + + /// + /// Reads one client frame. Returns the decoded text of the first text frame seen, or null + /// once the peer closes. Control frames other than Close are skipped. + /// + private async Task ReadTextFrameAsync(NetworkStream stream) + { + while (true) + { + byte[] head = new byte[2]; + if (!await ReadExactAsync(stream, head, 2).ConfigureAwait(false)) + { + return null; + } + + int opcode = head[0] & 0x0F; + bool masked = (head[1] & 0x80) != 0; + long length = head[1] & 0x7F; + + if (length == 126) + { + byte[] extended = new byte[2]; + if (!await ReadExactAsync(stream, extended, 2).ConfigureAwait(false)) + { + return null; + } + + length = BinaryPrimitives.ReadUInt16BigEndian(extended); + } + else if (length == 127) + { + byte[] extended = new byte[8]; + if (!await ReadExactAsync(stream, extended, 8).ConfigureAwait(false)) + { + return null; + } + + length = (long)BinaryPrimitives.ReadUInt64BigEndian(extended); + } + + byte[] mask = new byte[4]; + if (masked && !await ReadExactAsync(stream, mask, 4).ConfigureAwait(false)) + { + return null; + } + + byte[] payload = new byte[length]; + if (length > 0 && !await ReadExactAsync(stream, payload, (int)length).ConfigureAwait(false)) + { + return null; + } + + if (masked) + { + for (int i = 0; i < payload.Length; i++) + { + payload[i] ^= mask[i % 4]; + } + } + + if (opcode == 0x8) + { + return null; + } + + if (opcode == 0x1 || opcode == 0x2) + { + return Encoding.UTF8.GetString(payload); + } + } + } + + private async Task ReadExactAsync(NetworkStream stream, byte[] buffer, int count) + { + int read = 0; + + while (read < count) + { + int chunk = await stream.ReadAsync(buffer.AsMemory(read, count - read), _cts.Token) + .ConfigureAwait(false); + if (chunk == 0) + { + return false; + } + + read += chunk; + } + + return true; + } + + private async Task ReadUntilHeadersEndAsync(NetworkStream stream) + { + byte[] buffer = new byte[4096]; + StringBuilder request = new StringBuilder(); + + while (true) + { + int read = await stream.ReadAsync(buffer, _cts.Token).ConfigureAwait(false); + if (read == 0) + { + break; + } + + request.Append(Encoding.ASCII.GetString(buffer, 0, read)); + + if (request.ToString().Contains("\r\n\r\n", StringComparison.Ordinal)) + { + break; + } + } + + return request.ToString(); + } + + public void Dispose() + { + try + { + _cts.Cancel(); + } + catch + { + // best effort + } + + try + { + _listener.Stop(); + } + catch + { + // best effort + } + + _cts.Dispose(); + } + } +} diff --git a/Tests/Xrpl.Tests/Client/TestURequestManagerCancellation.cs b/Tests/Xrpl.Tests/Client/TestURequestManagerCancellation.cs new file mode 100644 index 00000000..fb4fc8c7 --- /dev/null +++ b/Tests/Xrpl.Tests/Client/TestURequestManagerCancellation.cs @@ -0,0 +1,102 @@ +using Microsoft.VisualStudio.TestTools.UnitTesting; + +using System; +using System.Collections.Concurrent; +using System.Collections.Generic; +using System.Reflection; +using System.Threading; +using System.Threading.Tasks; + +using Xrpl.Client; +using Xrpl.Models.Methods; + +using Timer = System.Timers.Timer; + +namespace Xrpl.Tests.ClientLib +{ + /// + /// A token that is already cancelled runs its registration callback inline, so the request is + /// rejected in the middle of being built — before its timeout timer exists. These tests pin + /// that nothing is left behind on that path: no timer in the timeout map, no pending promise. + /// + [TestClass] + public class TestURequestManagerCancellation + { + private static ConcurrentDictionary Timeouts(RequestManager manager) + { + FieldInfo field = typeof(RequestManager).GetField( + "timeoutsAwaitingResponse", + BindingFlags.Instance | BindingFlags.NonPublic); + + Assert.IsNotNull(field, "timeoutsAwaitingResponse is gone - update this test"); + return (ConcurrentDictionary)field.GetValue(manager); + } + + private static ConcurrentDictionary Promises(RequestManager manager) + { + FieldInfo field = typeof(RequestManager).GetField( + "promisesAwaitingResponse", + BindingFlags.Instance | BindingFlags.NonPublic); + + Assert.IsNotNull(field, "promisesAwaitingResponse is gone - update this test"); + return (ConcurrentDictionary)field.GetValue(manager); + } + + [TestMethod] + public async Task TestUCancelledTokenLeavesNoTimerBehindOnCreateRequest() + { + RequestManager manager = new RequestManager(); + using CancellationTokenSource cts = new CancellationTokenSource(); + cts.Cancel(); + + Dictionary request = new Dictionary { ["command"] = "ping" }; + RequestManager.XrplRequest created = manager.CreateRequest( + request, + TimeSpan.FromSeconds(30), + cts.Token); + + await Assert.ThrowsExactlyAsync(() => created.Promise); + + Assert.AreEqual(0, Timeouts(manager).Count, "timeout timer outlived the cancelled request"); + Assert.AreEqual(0, Promises(manager).Count, "promise outlived the cancelled request"); + } + + [TestMethod] + public async Task TestUCancelledTokenLeavesNoTimerBehindOnCreateGRequest() + { + RequestManager manager = new RequestManager(); + using CancellationTokenSource cts = new CancellationTokenSource(); + cts.Cancel(); + + RequestManager.XrplGRequest created = manager.CreateGRequest( + new PingRequest(), + TimeSpan.FromSeconds(30), + cts.Token); + + await Assert.ThrowsExactlyAsync(() => created.Promise); + + Assert.AreEqual(0, Timeouts(manager).Count, "timeout timer outlived the cancelled request"); + Assert.AreEqual(0, Promises(manager).Count, "promise outlived the cancelled request"); + } + + /// + /// The plain path still registers a timer and still cleans it up once the request finishes, + /// so the guard above cannot pass by never registering one in the first place. + /// + [TestMethod] + public void TestULiveRequestRegistersAndThenReleasesItsTimer() + { + RequestManager manager = new RequestManager(); + + Dictionary request = new Dictionary { ["command"] = "ping" }; + RequestManager.XrplRequest created = manager.CreateRequest(request, TimeSpan.FromSeconds(30)); + + Assert.AreEqual(1, Timeouts(manager).Count, "a live request must arm its timeout"); + + manager.Reject(created.Id, new OperationCanceledException("done")); + + Assert.AreEqual(0, Timeouts(manager).Count, "completing a request must release its timeout"); + Assert.AreEqual(0, Promises(manager).Count); + } + } +} diff --git a/Tests/Xrpl.Tests/Client/TestUWebSocketMessageAssembly.cs b/Tests/Xrpl.Tests/Client/TestUWebSocketMessageAssembly.cs new file mode 100644 index 00000000..4a4eee72 --- /dev/null +++ b/Tests/Xrpl.Tests/Client/TestUWebSocketMessageAssembly.cs @@ -0,0 +1,138 @@ +using Microsoft.VisualStudio.TestTools.UnitTesting; + +using System; +using System.Collections.Generic; +using System.Threading; +using System.Threading.Tasks; + +using Xrpl.Client; + +namespace Xrpl.Tests.ClientLib +{ + /// + /// Regression coverage for assembling a WebSocket message that arrives as several receive + /// chunks. The assembly buffer is reused for the life of the connection, so the tests check + /// both that every message comes out byte-exact and that the cost per message does not grow + /// with the number of chunks it was split into. + /// + // GC.GetTotalAllocatedBytes is process-wide, so the allocation assertion below would pick up + // whatever other test classes allocate alongside it. + [TestClass] + [DoNotParallelize] + public class TestUWebSocketMessageAssembly + { + private const int WaitSeconds = 120; + + [TestMethod] + public async Task TestUMultiChunkMessageArrivesIntact() + { + // Deliberately larger than the client's 1 MiB receive buffer and split far more finely + // than the buffer would split it on its own. + const int PayloadBytes = 3 * 1024 * 1024; + + using BulkMessageServer server = new BulkMessageServer(4, PayloadBytes, fragments: 96); + IReadOnlyList messages = await ReceiveAsync(server, 4).ConfigureAwait(false); + + Assert.AreEqual(4, messages.Count); + foreach (string message in messages) + { + Assert.AreEqual(server.PayloadText, message); + } + } + + [TestMethod] + public async Task TestUShortMessageAfterLongOneIsNotPaddedWithStaleBytes() + { + const int PayloadBytes = 2 * 1024 * 1024; + int[] lengthCycle = { PayloadBytes, PayloadBytes / 8, PayloadBytes / 2, 1024 }; + + using BulkMessageServer server = new BulkMessageServer( + messageCount: 12, + payloadBytes: PayloadBytes, + fragments: 16, + lengthCycle: lengthCycle); + + IReadOnlyList messages = await ReceiveAsync(server, 12).ConfigureAwait(false); + + Assert.AreEqual(12, messages.Count); + for (int i = 0; i < messages.Count; i++) + { + int expectedLength = lengthCycle[i % lengthCycle.Length]; + Assert.AreEqual(expectedLength, messages[i].Length, $"message {i} has the wrong length"); + Assert.AreEqual(server.PayloadText.Substring(0, expectedLength), messages[i], + $"message {i} does not match the expected prefix"); + } + } + + /// + /// Guards the shape of the fix: assembly used to be quadratic in the number of chunks, so + /// allocation per message grew with the split. The bound is deliberately loose — the point + /// is that a 64-way split must not cost an order of magnitude more than the payload. + /// + [TestMethod] + public async Task TestUAssemblyAllocationDoesNotGrowWithChunkCount() + { + const int MessageCount = 200; + const int PayloadBytes = 1024 * 1024; + + // Floor per message is the exact-sized byte[] plus the UTF-16 string handed to the + // callback, i.e. about 3x the payload. Quadratic assembly at 64 chunks cost ~34x. + const double AllowedTimesPayload = 12.0; + + using BulkMessageServer server = new BulkMessageServer(MessageCount, PayloadBytes, fragments: 64); + + GC.Collect(2, GCCollectionMode.Forced, blocking: true); + GC.WaitForPendingFinalizers(); + GC.Collect(2, GCCollectionMode.Forced, blocking: true); + + long allocatedBefore = GC.GetTotalAllocatedBytes(precise: true); + IReadOnlyList messages = await ReceiveAsync(server, MessageCount).ConfigureAwait(false); + long allocated = GC.GetTotalAllocatedBytes(precise: true) - allocatedBefore; + + Assert.AreEqual(MessageCount, messages.Count); + + double perMessage = allocated / (double)MessageCount / PayloadBytes; + Assert.IsTrue( + perMessage < AllowedTimesPayload, + $"allocated {perMessage:F1}x payload per message, expected below {AllowedTimesPayload:F1}x"); + } + + private static async Task> ReceiveAsync(BulkMessageServer server, int expected) + { + List messages = new List(expected); + TaskCompletionSource done = + new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + + WebSocketClient client = WebSocketClient.Create(server.Url); + client.OnMessageReceived((message, _) => + { + messages.Add(message); + if (messages.Count >= expected) + { + done.TrySetResult(true); + } + + return Task.CompletedTask; + }); + + try + { + await client.Connect().ConfigureAwait(false); + + // The server holds off until the client speaks, so nothing is missed. + client.SendMessage("go"); + + Task completed = await Task.WhenAny(done.Task, Task.Delay(TimeSpan.FromSeconds(WaitSeconds))) + .ConfigureAwait(false); + Assert.AreSame(done.Task, completed, $"only {messages.Count} of {expected} messages arrived"); + } + finally + { + client.CancelIntentionally(); + client.Dispose(); + } + + return messages; + } + } +} diff --git a/Xrpl/Client/IXrplClient.cs b/Xrpl/Client/IXrplClient.cs index 0a87af1b..90633b2f 100644 --- a/Xrpl/Client/IXrplClient.cs +++ b/Xrpl/Client/IXrplClient.cs @@ -1,5 +1,4 @@ using System; -using System.Collections.Concurrent; using System.Collections.Generic; using System.Text.Json; using System.Threading; @@ -487,8 +486,6 @@ public class ClientOptions : ConnectionOptions ///// Current web socket client state //public WebSocketState SocketState => client.State; - private readonly ConcurrentDictionary tasks; - public XrplClient(string server, ClientOptions? options = null) { diff --git a/Xrpl/Client/RequestManager.cs b/Xrpl/Client/RequestManager.cs index be85d284..2a18e1be 100644 --- a/Xrpl/Client/RequestManager.cs +++ b/Xrpl/Client/RequestManager.cs @@ -4,6 +4,7 @@ using System.Collections.Concurrent; using System.Collections.Generic; using System.Diagnostics; +using System.Reflection; using System.Text.Json; using System.Text.Json.Nodes; using System.Threading; @@ -65,17 +66,16 @@ public void Resolve(Guid id, BaseResponse response) if (!promisesAwaitingResponse.TryGetValue(id, out var taskInfo) || taskInfo == null) { Debug.WriteLine($"Resolve called for non-existent promise {id} (likely already cancelled/timed out)"); + DisposeTimeout(id); return; } - if (timeoutsAwaitingResponse.TryRemove(id, out var timer)) - timer.Stop(); + DisposeTimeout(id); try { - var deserialized = JsonSerializer.Deserialize(response.Result?.ToString() ?? "{}", taskInfo.Type, serializerOptions); - var setResult = taskInfo.TaskCompletionResult.GetType().GetMethod("TrySetResult"); - setResult.Invoke(taskInfo.TaskCompletionResult, new[] { deserialized }); + object deserialized = JsonSerializer.Deserialize(response.Result?.ToString() ?? "{}", taskInfo.Type, serializerOptions); + CompleteWithResult(taskInfo, deserialized); this.DeletePromise(id, taskInfo); } catch (Exception ex) @@ -97,34 +97,88 @@ public void Reject(Guid id, T error) where T : Exception if (!promisesAwaitingResponse.TryGetValue(id, out var taskInfo) || taskInfo == null) { Debug.WriteLine($"Reject called for non-existent promise {id} (likely already resolved)"); + + // A timer registered after its request had already finished has no other chance of + // being cleaned up: this is the Elapsed callback of exactly such a timer. + DisposeTimeout(id); return; } - if (timeoutsAwaitingResponse.TryRemove(id, out var timer)) - timer.Stop(); - var setException = taskInfo.TaskCompletionResult.GetType().GetMethod("TrySetException", new Type[] { typeof(Exception) }, null); - setException.Invoke(taskInfo.TaskCompletionResult, new[] { error }); - + + DisposeTimeout(id); + CompleteWithException(taskInfo, error); + // Observe the exception to prevent UnobservedTaskException in consuming apps // This is critical for MAUI/mobile apps that have global exception handlers - ObserveTaskException(taskInfo.TaskCompletionResult); + ObserveTaskException(taskInfo); this.DeletePromise(id, taskInfo); } + /// + /// Removes the timeout timer of and disposes it. + /// Dispose, not Stop: is a finalizable Component, and a stopped but + /// undisposed one per request piles up on the finalization queue over a long paged run. + /// + private void DisposeTimeout(Guid id) + { + if (timeoutsAwaitingResponse.TryRemove(id, out Timer timer)) + { + timer.Dispose(); + } + } + + /// + /// Completes the pending request with a deserialized result, using the typed delegate + /// captured when the request was created and falling back to reflection for + /// instances built outside this manager. + /// + private static void CompleteWithResult(TaskInfo taskInfo, object result) + { + if (taskInfo.SetResult is not null) + { + taskInfo.SetResult(result); + return; + } + + MethodInfo setResult = taskInfo.TaskCompletionResult.GetType().GetMethod("TrySetResult"); + setResult.Invoke(taskInfo.TaskCompletionResult, new[] { result }); + } + + /// + /// Faults the pending request. See for the fallback rules. + /// + private static void CompleteWithException(TaskInfo taskInfo, Exception error) + { + if (taskInfo.SetException is not null) + { + taskInfo.SetException(error); + return; + } + + MethodInfo setException = taskInfo.TaskCompletionResult.GetType() + .GetMethod("TrySetException", new Type[] { typeof(Exception) }, null); + setException.Invoke(taskInfo.TaskCompletionResult, new object[] { error }); + } + /// /// Observes the exception on a TaskCompletionSource's Task to prevent UnobservedTaskException. /// When a Task faults but is never awaited, .NET raises UnobservedTaskException event. /// By adding a ContinueWith that reads the exception, we mark it as "observed". /// - private void ObserveTaskException(object taskCompletionSource) + private void ObserveTaskException(TaskInfo taskInfo) { try { - // Get the Task property from TaskCompletionSource - var taskProperty = taskCompletionSource.GetType().GetProperty("Task"); - if (taskProperty == null) return; - - var task = taskProperty.GetValue(taskCompletionSource) as Task; + Task task = taskInfo.CompletionTask; + if (task == null) + { + // Externally built TaskInfo: fall back to reading the Task property reflectively. + PropertyInfo taskProperty = taskInfo.TaskCompletionResult.GetType().GetProperty("Task"); + if (taskProperty == null) return; + + task = taskProperty.GetValue(taskInfo.TaskCompletionResult) as Task; + } + if (task == null) return; // Add a continuation that observes the exception (reads it to mark as handled) @@ -214,6 +268,9 @@ public XrplGRequest CreateGRequest( TaskInfo taskInfo = new TaskInfo(); taskInfo.TaskId = newId; taskInfo.TaskCompletionResult = task; + taskInfo.SetResult = result => task.TrySetResult(result); + taskInfo.SetException = error => task.TrySetException(error); + taskInfo.CompletionTask = task.Task; taskInfo.RemoveUponCompletion = true; taskInfo.Type = typeof(T); @@ -236,6 +293,14 @@ public XrplGRequest CreateGRequest( } }); taskInfo.CancellationRegistration = registration; + + // An already cancelled token runs its callback inline, so Reject completed and + // removed the promise before the registration was stored — nothing would ever + // dispose it. + if (!promisesAwaitingResponse.ContainsKey(newId)) + { + _ = registration.DisposeAsync(); + } } if (timeout != System.Threading.Timeout.InfiniteTimeSpan) @@ -255,6 +320,15 @@ public XrplGRequest CreateGRequest( }; timer.Start(); timeoutsAwaitingResponse.TryAdd(newId, timer); + + // Same inline-cancellation case: the request may already be finished, and the + // Reject that finished it ran before this timer existed, so it had nothing to + // remove. Whatever is registered under this id now belongs to a request that is + // already gone. + if (!promisesAwaitingResponse.ContainsKey(newId)) + { + DisposeTimeout(newId); + } } return new XrplGRequest() @@ -291,6 +365,9 @@ public XrplRequest CreateRequest( TaskInfo taskInfo = new TaskInfo(); taskInfo.TaskId = newId; taskInfo.TaskCompletionResult = task; + taskInfo.SetResult = result => task.TrySetResult((Dictionary)result); + taskInfo.SetException = error => task.TrySetException(error); + taskInfo.CompletionTask = task.Task; taskInfo.RemoveUponCompletion = true; taskInfo.Type = typeof(Dictionary); @@ -313,6 +390,14 @@ public XrplRequest CreateRequest( } }); taskInfo.CancellationRegistration = registration; + + // An already cancelled token runs its callback inline, so Reject completed and + // removed the promise before the registration was stored — nothing would ever + // dispose it. + if (!promisesAwaitingResponse.ContainsKey(newId)) + { + _ = registration.DisposeAsync(); + } } if (timeout != System.Threading.Timeout.InfiniteTimeSpan) @@ -332,6 +417,15 @@ public XrplRequest CreateRequest( }; timer.Start(); timeoutsAwaitingResponse.TryAdd(newId, timer); + + // Same inline-cancellation case: the request may already be finished, and the + // Reject that finished it ran before this timer existed, so it had nothing to + // remove. Whatever is registered under this id now belongs to a request that is + // already gone. + if (!promisesAwaitingResponse.ContainsKey(newId)) + { + DisposeTimeout(newId); + } } return new XrplRequest() diff --git a/Xrpl/Client/TaskInfo.cs b/Xrpl/Client/TaskInfo.cs index ba478f14..6b4db338 100644 --- a/Xrpl/Client/TaskInfo.cs +++ b/Xrpl/Client/TaskInfo.cs @@ -1,5 +1,6 @@ using System; using System.Threading; +using System.Threading.Tasks; namespace Xrpl.Client { @@ -11,6 +12,25 @@ public class TaskInfo public object TaskCompletionResult { get; set; } + /// + /// Completes with a deserialized result. Set when the + /// request is created, so the response path does not have to reach for the strongly typed + /// TrySetResult through reflection. Null for externally built instances. + /// + public Func SetResult { get; set; } + + /// + /// Faults . Set when the request is created; null for + /// externally built instances. + /// + public Func SetException { get; set; } + + /// + /// The task behind , used to observe faults without + /// reflection. Null for externally built instances. + /// + public Task CompletionTask { get; set; } + public bool RemoveUponCompletion { get; set; } public CancellationTokenRegistration? CancellationRegistration { get; set; } diff --git a/Xrpl/Client/WebSocketClient.cs b/Xrpl/Client/WebSocketClient.cs index dd8636e8..8a0c885d 100644 --- a/Xrpl/Client/WebSocketClient.cs +++ b/Xrpl/Client/WebSocketClient.cs @@ -1,10 +1,10 @@  using System; +using System.Buffers; using System.Collections.Generic; using System.Diagnostics; using System.IO; -using System.Linq; using System.Net.WebSockets; using System.Text; using System.Threading; @@ -476,43 +476,66 @@ await Task.WhenAny( private async Task ReceiveLoopAsync() { - byte[] buffer = new byte[ReceiveChunkSize]; + // One receive buffer per connection, rented rather than allocated: at ReceiveChunkSize + // it lands on the large object heap, so a reconnect loop would otherwise burn a fresh + // uncompactable megabyte per session. Returned in the finally below, which only runs + // once the loop has stopped awaiting the socket. + byte[] buffer = ArrayPool.Shared.Rent(ReceiveChunkSize); + + // Scratch buffer for messages that arrive in more than one chunk. It grows to the + // largest message seen on this connection and is then reused, so steady-state assembly + // costs nothing beyond the single exact-sized array handed to the callbacks. + byte[]? assemblyBuffer = null; try { // Continue receiving while Open OR CloseSent (waiting for server's Close frame after CloseOutputAsync) - while (_ws != null && - (_ws.State == WebSocketState.Open || _ws.State == WebSocketState.CloseSent) && + while (_ws != null && + (_ws.State == WebSocketState.Open || _ws.State == WebSocketState.CloseSent) && !_cancellationToken.IsCancellationRequested) { - byte[] byteResult = Array.Empty(); + byte[]? completeMessage = null; + int assembledLength = 0; WebSocketReceiveResult result; - bool timedOut = false; do { - result = await _ws.ReceiveAsync(new ArraySegment(buffer), _cancellationToken).ConfigureAwait(false); + result = await _ws.ReceiveAsync(new ArraySegment(buffer, 0, ReceiveChunkSize), _cancellationToken).ConfigureAwait(false); if (result.MessageType == WebSocketMessageType.Close) { WebSocketCloseStatus? closeStatus = result.CloseStatus; string? closeDescription = result.CloseStatusDescription; - + _onClosed?.Invoke(this); await CallOnDisconnectedAsync(closeStatus, closeDescription).ConfigureAwait(false); return; } - else + else if (result.EndOfMessage && assembledLength == 0) + { + // Whole message arrived in a single chunk - copy it out directly. + completeMessage = new byte[result.Count]; + Buffer.BlockCopy(buffer, 0, completeMessage, 0, result.Count); + } + else if (result.Count > 0) { - byteResult = byteResult.Concat(buffer.Take(result.Count)).ToArray(); + EnsureAssemblyCapacity(ref assemblyBuffer, assembledLength + result.Count, assembledLength); + Buffer.BlockCopy(buffer, 0, assemblyBuffer, assembledLength, result.Count); + assembledLength += result.Count; } } while (!result.EndOfMessage); - if (timedOut) - continue; + if (completeMessage == null) + { + completeMessage = new byte[assembledLength]; + if (assembledLength > 0) + { + Buffer.BlockCopy(assemblyBuffer!, 0, completeMessage, 0, assembledLength); + } + } - CallOnMessage(byteResult); + CallOnMessage(completeMessage); } } catch (OperationCanceledException) when (_isIntentionalDisconnect || _cancellationToken.IsCancellationRequested || IsDisposed) @@ -611,6 +634,37 @@ private async Task ReceiveLoopAsync() _onConnectionError?.Invoke(ex, this); await CallOnDisconnectedAsync(WebSocketCloseStatus.EndpointUnavailable, "Unknown error: " + ex.Message).ConfigureAwait(false); } + finally + { + ArrayPool.Shared.Return(buffer); + } + } + + /// + /// Grows so it can hold + /// bytes, doubling the current capacity and carrying over the first + /// bytes already assembled. + /// + private static void EnsureAssemblyCapacity(ref byte[]? assemblyBuffer, int requiredLength, int preserveLength) + { + if (assemblyBuffer != null && assemblyBuffer.Length >= requiredLength) + { + return; + } + + int capacity = assemblyBuffer?.Length ?? ReceiveChunkSize; + while (capacity < requiredLength) + { + capacity = capacity <= Array.MaxLength / 2 ? capacity * 2 : requiredLength; + } + + byte[] grown = new byte[capacity]; + if (preserveLength > 0) + { + Buffer.BlockCopy(assemblyBuffer!, 0, grown, 0, preserveLength); + } + + assemblyBuffer = grown; } private void CallOnMessage(byte[] result) From 5f1d1799350a9788ec365c6aa8036d93486f9a2c Mon Sep 17 00:00:00 2001 From: Aleksandr <44946855+Platonenkov@users.noreply.github.com> Date: Thu, 13 Aug 2026 19:57:38 +0000 Subject: [PATCH 3/3] =?UTF-8?q?perf(client):=20scratch-=D0=B1=D1=83=D1=84?= =?UTF-8?q?=D0=B5=D1=80=20=D1=81=D0=B1=D0=BE=D1=80=D0=BA=D0=B8=20=D1=82?= =?UTF-8?q?=D0=BE=D0=B6=D0=B5=20=D0=B8=D0=B7=20=D0=BF=D1=83=D0=BB=D0=B0=20?= =?UTF-8?q?+=20=D0=B7=D0=B0=D0=BC=D0=B5=D1=87=D0=B0=D0=BD=D0=B8=D1=8F=20Co?= =?UTF-8?q?deRabbit=20=D0=BA=20PR=20#88=20(#90)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * perf(client): scratch-буфер сборки тоже из пула + замечания CodeRabbit к PR #88 - WebSocketClient: assemblyBuffer брался обычной аллокацией, хотя буфер приёма уже шёл из ArrayPool. Замерено на .NET 10: общий пул возвращает тот же массив после Return и на 2, и на 4 МиБ, так что это снимает последнюю LOH-аллокацию на соединение, а не добавляет косвенности. Старый массив возвращается в пул при росте, текущий — в том же finally, что и буфер приёма - тестовые WS-серверы: framing, хендшейк, приём заголовков и Dispose вынесены в WebSocketTestServerBase. Дублирование здесь уже один раз стрельнуло — кламп offset существовал в BulkMessageServer и отсутствовал в PagedResponseServer, что и поймал предыдущий проход ревью - BulkMessageServer проверяет lengthCycle в конструкторе: цикл отправки идёт отдельной задачей, поэтому длина больше payload вылезала не ошибкой аргумента, а таймаутом приёма через две минуты - CHANGES: «the other six converters» перечисляло девять. Само число верное — WithoutConverter зовут ровно шесть типов; три node-конвертера его не зовут и в эту шестёрку не входят, они проверялись по другому признаку. Формулировка разведена, счёт не менялся * fix(tests): не терять ошибку сервера и не оставлять слушателя при отказе конструктора По итогам селф-ревью рефакторинга харнесса. - PagedResponseServer раньше глушил IOException/SocketException/ObjectDisposedException сам, а после выноса в базу под общий catch попадали уже любые исключения и молча пропадали. База теперь запоминает их в Fault, BulkMessageServer по-прежнему дополнительно роняет в SendCompleted, а тест сборки показывает Fault в сообщении об ошибке — иначе поломка сервера выглядела бы как таймаут приёма на стороне клиента - конструктор базы уже поднимает слушателя к моменту, когда отрабатывает проверка lengthCycle в наследнике, так что бросок оставлял открытый сокет и неосвобождённый CTS без единого владельца. Перед throw вызывается Dispose * fix(tests): не читать Token у уже освобождённого CTS в тестовом WS-сервере Замечания CodeRabbit к PR #90. - Dispose звал Cancel и сразу Dispose, не дожидаясь цикла приёма, а тот и DrainAsync читают Token на каждой итерации. Чтение CancellationTokenSource.Token после Dispose бросает ObjectDisposedException, и она уходила в общий catch — то есть в Fault попадал артефакт разбора стенда вместо настоящей ошибки сервера, а у BulkMessageServer ещё и в SendCompleted. Токен теперь снимается один раз в конструкторе: Dispose всегда отменяет перед освобождением, а по уже отменённому токену ожидания завершаются, не обращаясь к источнику - ReadUntilHeadersEndAsync может прочитать за "\r\n\r\n" и остаток молча теряет. Живого клиента это не задевает — фрейм нельзя слать до 101, — но база теперь общая, поэтому ограничение описано в док-комментарии --- CHANGES.md | 4 +- Tests/Xrpl.Tests/Client/BulkMessageServer.cs | 207 ++++------------ .../Xrpl.Tests/Client/PagedResponseServer.cs | 188 ++------------- .../Client/TestUWebSocketMessageAssembly.cs | 5 +- .../Client/WebSocketTestServerBase.cs | 227 ++++++++++++++++++ Xrpl/Client/WebSocketClient.cs | 18 +- 6 files changed, 316 insertions(+), 333 deletions(-) create mode 100644 Tests/Xrpl.Tests/Client/WebSocketTestServerBase.cs diff --git a/CHANGES.md b/CHANGES.md index e1db87e1..ad9e1f5c 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -5,9 +5,9 @@ * **Fix infinite recursion in `LONFTokenConverter.Write` — the metadata of an NFT transaction could not be serialized at all** — regression introduced in 10.3.0.0 with the `Newtonsoft.Json` → `System.Text.Json` migration; affects every release from 10.3.0.0 on. `JsonSerializer.Serialize(tx.Meta)` threw `JsonException: A possible object cycle was detected` for any transaction whose `AffectedNodes` contain an `NFTokenPage`, which is every `NFTokenMint`, `NFTokenBurn`, `NFTokenAcceptOffer` and `NFTokenModify` that touched a page. Verified against mainnet on all six NFT transaction types — the four above failed, `NFTokenCreateOffer` and `NFTokenCancelOffer` (no page in their metadata) went through: * The converter broke its own recursion the way the other polymorphic converters do — strip itself from `options.Converters` via `JsonSerializerOptionsCache.WithoutConverter` and re-enter the serializer. That works only for a converter that is *registered in the list*. `LONFTokenConverter` is declared as a `[JsonConverter]` **attribute on the `NFToken` type itself** (`LONFTokenPage.cs`), and a converter attached to a type outranks the options list, so System.Text.Json handed the value straight back to `Write` no matter what the list looked like. The frame repeated until the writer hit `MaxDepth`. Raising `MaxDepth` is not a workaround: at 64 and 128 it is a catchable `JsonException`, at 256 the stack overflows and the process dies * `NFToken` has two fields, so `Write` now emits them directly instead of delegating. The wire shape is unchanged — `{"NFToken":{"NFTokenID":"…","URI":"…"}}`, the envelope `Read` already looks for — and the documented null behaviour is preserved by honouring `options.DefaultIgnoreCondition` rather than hard-coding one: `XrplJsonOptions.Default` (`WhenWritingNull`) omits a null `URI`, plain options keep it as `null` - * The other six converters that call `WithoutConverter` were audited against the same two conditions — declared as a type-level attribute **and** re-serializing that same declared type. None hit both. `LOConverter` is registered in the options list (its one attribute use is property-level) and writes the concrete runtime type; `GenericStringConverter`, `MetaBinaryConverter`, `LedgerBinaryConverter` and `TransactionRequestConverter` are only ever attached to properties; the three node converters are type-level but serialize a *different* class (`value.NewFields.GetType()`). `TransactionResponseConverter` is the one other type-level case, and the same trap was already defused there by the `TransactionResponseUnknown` sentinel, so that no value ever carries the annotated type at runtime. Nothing else was changed + * The six other converter types that call `WithoutConverter` — `LOConverter`, `GenericStringConverter`, `MetaBinaryConverter`, `LedgerBinaryConverter`, `TransactionRequestConverter` and `TransactionResponseConverter` — were audited against the same two conditions — declared as a type-level attribute **and** re-serializing that same declared type. None hit both. `LOConverter` is registered in the options list (its one attribute use is property-level) and writes the concrete runtime type; `GenericStringConverter`, `MetaBinaryConverter`, `LedgerBinaryConverter` and `TransactionRequestConverter` are only ever attached to properties. The three node converters (`CreatedNodeConverter`, `ModifiedNodeConverter`, `DeletedNodeConverter`) do not call `WithoutConverter` at all and so are not among those six, but they are type-level and were checked for the same trap anyway: they serialize a *different* class (`value.NewFields.GetType()`). `TransactionResponseConverter` is the one other type-level case, and the same trap was already defused there by the `TransactionResponseUnknown` sentinel, so that no value ever carries the annotated type at runtime. Nothing else was changed * `TestULONFTokenConverter` had `Read` coverage only, which is how the bug survived. It now pins the written shape, both round trips (`URI` set and null), null handling under `XrplJsonOptions.Default` and under plain options, a multi-token `NFTokenPage`, and — the regression test proper — serializing a `Meta` carrying an `NFTokenPage` in `CreatedNode.NewFields`, `ModifiedNode.FinalFields`, `ModifiedNode.PreviousFields` and `DeletedNode.FinalFields`, since the page can arrive in any of them. All offline, on prepared JSON -* **WebSocket message assembly was quadratic in the number of receive chunks** — `ReceiveLoopAsync` grew a multi-chunk message with `byteResult = byteResult.Concat(buffer.Take(result.Count)).ToArray()`. Every chunk allocated a fresh array the size of everything received so far and refilled it one byte at a time through a LINQ enumerator, so a message split into *k* chunks copied roughly `k/2` times its own length; every intermediate array was well past the 85 KB threshold and therefore landed on the uncompacted large object heap. `ledger_data` at `limit=2048` is a few megabytes and arrives in dozens of chunks over a real link, which is exactly where the cost concentrates. Chunks are now `Buffer.BlockCopy`-ed into a scratch buffer that grows to the largest message on the connection and is reused from then on; a message that arrives whole in one chunk skips the scratch entirely, and the receive buffer itself is rented from `ArrayPool` rather than allocated per connection. Measured on a local fragmenting WebSocket server, 300 messages of 2 MiB, allocation per message: 3.50x payload at one chunk, 6.50x at eight, 18.52x at thirty-two — now a flat 3.01x, which is the floor (the exact-sized `byte[]` plus the UTF-16 string handed to the callback). Through the full client stack, a 3000-page `ledger_data` crawl with each page arriving in 32 chunks and a consumer retaining every object: 100.7 s → 55.8 s, 158.4 GiB → 67.3 GiB allocated, 891 → 398 gen2 collections, and the last-decile-to-first-decile page time drops from 1.39x to 1.13x. `ReceiveChunkSize` was measured at 1 MiB and 64 KiB and left at 1 MiB — now that the buffer is pooled, shrinking it changed nothing outside run-to-run noise. `TestUWebSocketMessageAssembly` pins the byte-exactness of a 96-chunk message, that a short message after a long one picks up no stale bytes from the reused buffer, and that per-message allocation stays under 12x payload at 64 chunks (34.5x before the fix). A dead `timedOut` local, declared and tested but never assigned since it appeared, is gone +* **WebSocket message assembly was quadratic in the number of receive chunks** — `ReceiveLoopAsync` grew a multi-chunk message with `byteResult = byteResult.Concat(buffer.Take(result.Count)).ToArray()`. Every chunk allocated a fresh array the size of everything received so far and refilled it one byte at a time through a LINQ enumerator, so a message split into *k* chunks copied roughly `k/2` times its own length; every intermediate array was well past the 85 KB threshold and therefore landed on the uncompacted large object heap. `ledger_data` at `limit=2048` is a few megabytes and arrives in dozens of chunks over a real link, which is exactly where the cost concentrates. Chunks are now `Buffer.BlockCopy`-ed into a scratch buffer that grows to the largest message on the connection and is reused from then on; a message that arrives whole in one chunk skips the scratch entirely, and both the receive buffer and that scratch buffer are rented from `ArrayPool` rather than allocated per connection (measured on .NET 10: the shared pool does hand back the same multi-megabyte array after a return, so this is a real saving and not just indirection). Measured on a local fragmenting WebSocket server, 300 messages of 2 MiB, allocation per message: 3.50x payload at one chunk, 6.50x at eight, 18.52x at thirty-two — now a flat 3.01x, which is the floor (the exact-sized `byte[]` plus the UTF-16 string handed to the callback). Through the full client stack, a 3000-page `ledger_data` crawl with each page arriving in 32 chunks and a consumer retaining every object: 100.7 s → 55.8 s, 158.4 GiB → 67.3 GiB allocated, 891 → 398 gen2 collections, and the last-decile-to-first-decile page time drops from 1.39x to 1.13x. `ReceiveChunkSize` was measured at 1 MiB and 64 KiB and left at 1 MiB — now that the buffer is pooled, shrinking it changed nothing outside run-to-run noise. `TestUWebSocketMessageAssembly` pins the byte-exactness of a 96-chunk message, that a short message after a long one picks up no stale bytes from the reused buffer, and that per-message allocation stays under 12x payload at 64 chunks (34.5x before the fix). A dead `timedOut` local, declared and tested but never assigned since it appeared, is gone * **Request timeout timers outlived their requests** — `RequestManager.Resolve`/`Reject` called `timer.Stop()`. `System.Timers.Timer` derives from `Component` and carries a finalizer, so every completed request left a finalizable object behind, each of them holding its request's serialized text alive through the `Elapsed` closure; over a long paged crawl that is thousands of them. `Dispose()` stops the timer and takes it off the finalization queue. A second, worse case sat next to it: a token that is **already cancelled** runs its `Register` callback inline, so `Reject` completed the request in the middle of the factory method — before the timeout timer existed and therefore with nothing to remove. The factory then registered the timer for a promise that was already gone, and when it fired, `Reject` took its missing-promise early return without removing it, so the entry stayed in `timeoutsAwaitingResponse` for the life of the process. The `CancellationTokenRegistration` leaked on the same path, its assignment to `TaskInfo` happening after `DeletePromise` had already run. Both factories now check whether the promise survived and clean up after themselves; timer removal moved into `DisposeTimeout`, which is also called on the early returns of `Resolve` and `Reject` and so closes the narrow race with a concurrent cancellation as well. `TestURequestManagerCancellation` pins that an already-cancelled token leaves neither timer nor promise behind in either factory, and that a live request still arms its timeout and releases it on completion * **Reflection on the per-response path is gone** — `Resolve`, `Reject` and `ObserveTaskException` reached for `TrySetResult`, `TrySetException` and `Task` through `GetType().GetMethod(...)` + `Invoke` on every single response. `TaskInfo` now carries typed `SetResult`/`SetException` delegates and the `CompletionTask` itself, wired when the request is created. The properties were added rather than substituted: `TaskInfo` is public, so instances built outside `RequestManager` keep the old reflective path * **Dead `tasks` field removed from `XrplClient`** — `private readonly ConcurrentDictionary tasks` was never assigned and never read, so it was permanently null; a leftover from when the client tracked pending requests itself, which `RequestManager` has done for a long time diff --git a/Tests/Xrpl.Tests/Client/BulkMessageServer.cs b/Tests/Xrpl.Tests/Client/BulkMessageServer.cs index 62445f30..c0653e89 100644 --- a/Tests/Xrpl.Tests/Client/BulkMessageServer.cs +++ b/Tests/Xrpl.Tests/Client/BulkMessageServer.cs @@ -1,25 +1,19 @@ using System; -using System.Net; using System.Net.Sockets; using System.Text; -using System.Threading; using System.Threading.Tasks; -using Xrpl.Tests.MockRippled; - namespace Xrpl.Tests { /// - /// Minimal WebSocket server that completes a handshake and then pushes a fixed number of - /// large text messages, each split into a controlled number of WebSocket continuation - /// frames. Fragmenting at the protocol level (rather than relying on how the socket happens - /// to slice the stream) makes the number of client-side receive chunks per message exact and - /// reproducible, which is what the assembly path is sensitive to. + /// WebSocket server that pushes a fixed number of large text messages once the client says go, + /// each split into a controlled number of WebSocket continuation frames. Fragmenting at the + /// protocol level (rather than relying on how the socket happens to slice the stream) makes the + /// number of client-side receive chunks per message exact and reproducible, which is what the + /// assembly path is sensitive to. /// - internal sealed class BulkMessageServer : IDisposable + internal sealed class BulkMessageServer : WebSocketTestServerBase { - private readonly TcpListener _listener; - private readonly CancellationTokenSource _cts = new(); private readonly int _messageCount; private readonly int _fragments; private readonly int[] _lengthCycle; @@ -45,17 +39,25 @@ public BulkMessageServer(int messageCount, int payloadBytes, int fragments, int[ _payload = BuildPayload(payloadBytes); _lengthCycle = lengthCycle is { Length: > 0 } ? lengthCycle : new[] { _payload.Length }; - _listener = new TcpListener(IPAddress.Loopback, 0); - _listener.Start(); - Port = ((IPEndPoint)_listener.LocalEndpoint).Port; - _ = AcceptAsync(); - } - - public int Port { get; } + // Checked here rather than left to fail mid-send: the send loop runs detached, so a bad + // length would only surface as the test's receive timeout minutes later. + foreach (int length in _lengthCycle) + { + if (length < 0 || length > _payload.Length) + { + // The base constructor has already opened the listener, and a constructor that + // throws leaves nobody to dispose it. + Dispose(); + throw new ArgumentOutOfRangeException( + nameof(lengthCycle), + $"length {length} must be between 0 and the payload length {_payload.Length}"); + } + } - public string Url => "ws://127.0.0.1:" + Port + "/"; + StartAccepting(); + } - /// Payload every message carries, as the client should see it. + /// Payload every message is a prefix of, as the client should see it. public string PayloadText => Encoding.UTF8.GetString(_payload); public int PayloadBytes => _payload.Length; @@ -96,67 +98,37 @@ private static byte[] BuildPayload(int payloadBytes) return Encoding.UTF8.GetBytes(builder.ToString(0, payloadBytes)); } - private async Task AcceptAsync() + protected override async Task ServeAsync(NetworkStream stream) { - try + // Wait for the client's go-ahead before pushing anything, so no message can land + // before the caller has opened its measurement window. + byte[] goAhead = new byte[256]; + if (await stream.ReadAsync(goAhead, Token).ConfigureAwait(false) == 0) { - using TcpClient client = await _listener.AcceptTcpClientAsync(_cts.Token).ConfigureAwait(false); - client.NoDelay = true; - NetworkStream stream = client.GetStream(); - - string request = await ReadUntilHeadersEndAsync(stream).ConfigureAwait(false); - string key = Helpers.GetHandshakeRequestKey(request); - byte[] response = Encoding.ASCII.GetBytes(Helpers.GetHandshakeResponse(Helpers.HashKey(key))); - await stream.WriteAsync(response, _cts.Token).ConfigureAwait(false); - await stream.FlushAsync(_cts.Token).ConfigureAwait(false); - - // Wait for the client's go-ahead before pushing anything, so no message can land - // before the caller has opened its measurement window. - byte[] goAhead = new byte[256]; - if (await stream.ReadAsync(goAhead, _cts.Token).ConfigureAwait(false) == 0) - { - _finished.TrySetResult(0); - return; - } - - // Drain and discard whatever the client sends afterwards (keep-alive pings, close - // frames) so its socket never blocks on a full send window. - _ = DrainAsync(stream); + _finished.TrySetResult(0); + return; + } - for (int i = 0; i < _messageCount; i++) - { - int messageLength = _lengthCycle[i % _lengthCycle.Length]; - int fragmentBytes = (messageLength + _fragments - 1) / _fragments; + // Drain and discard whatever the client sends afterwards (keep-alive pings, close + // frames) so its socket never blocks on a full send window. + _ = DrainAsync(stream); - for (int fragment = 0; fragment < _fragments; fragment++) - { - // Ceil division can leave trailing frames past the end; they are sent empty - // so the frame count stays exactly as requested. - int offset = Math.Min(fragment * fragmentBytes, messageLength); - int length = Math.Min(fragmentBytes, messageLength - offset); - bool isFirst = fragment == 0; - bool isLast = fragment == _fragments - 1; + for (int i = 0; i < _messageCount; i++) + { + int messageLength = _lengthCycle[i % _lengthCycle.Length]; + await WriteFragmentedMessageAsync(stream, _payload.AsMemory(0, messageLength), _fragments) + .ConfigureAwait(false); + } - await stream.WriteAsync(BuildFrameHeader(length, isFirst, isLast), _cts.Token) - .ConfigureAwait(false); - await stream.WriteAsync(_payload.AsMemory(offset, length), _cts.Token) - .ConfigureAwait(false); - await stream.FlushAsync(_cts.Token).ConfigureAwait(false); - } - } + _finished.TrySetResult(_messageCount); + } - _finished.TrySetResult(_messageCount); + protected override void OnCancelled() => _finished.TrySetCanceled(); - await Task.Delay(Timeout.InfiniteTimeSpan, _cts.Token).ConfigureAwait(false); - } - catch (OperationCanceledException) - { - _finished.TrySetCanceled(); - } - catch (Exception ex) - { - _finished.TrySetException(ex); - } + protected override void OnFaulted(Exception error) + { + base.OnFaulted(error); + _finished.TrySetException(error); } private async Task DrainAsync(NetworkStream stream) @@ -165,7 +137,7 @@ private async Task DrainAsync(NetworkStream stream) try { - while (await stream.ReadAsync(sink, _cts.Token).ConfigureAwait(false) > 0) + while (await stream.ReadAsync(sink, Token).ConfigureAwait(false) > 0) { } } @@ -174,88 +146,5 @@ private async Task DrainAsync(NetworkStream stream) // The connection going away is the normal end of this loop. } } - - /// - /// Unmasked server-to-client frame header. The first frame of a message carries the text - /// opcode (0x1), every following frame carries continuation (0x0); FIN is set on the last. - /// - private static byte[] BuildFrameHeader(int payloadLength, bool isFirst, bool isLast) - { - byte first = (byte)((isLast ? 0x80 : 0x00) | (isFirst ? 0x01 : 0x00)); - - if (payloadLength <= 125) - { - return new byte[] { first, (byte)payloadLength }; - } - - if (payloadLength <= ushort.MaxValue) - { - return new byte[] - { - first, - 126, - (byte)(payloadLength >> 8), - (byte)payloadLength - }; - } - - return new byte[] - { - first, - 127, - 0, 0, 0, 0, - (byte)(payloadLength >> 24), - (byte)(payloadLength >> 16), - (byte)(payloadLength >> 8), - (byte)payloadLength - }; - } - - private async Task ReadUntilHeadersEndAsync(NetworkStream stream) - { - byte[] buffer = new byte[4096]; - StringBuilder request = new StringBuilder(); - - while (true) - { - int read = await stream.ReadAsync(buffer, _cts.Token).ConfigureAwait(false); - if (read == 0) - { - break; - } - - request.Append(Encoding.ASCII.GetString(buffer, 0, read)); - - if (request.ToString().Contains("\r\n\r\n", StringComparison.Ordinal)) - { - break; - } - } - - return request.ToString(); - } - - public void Dispose() - { - try - { - _cts.Cancel(); - } - catch - { - // best effort - } - - try - { - _listener.Stop(); - } - catch - { - // best effort - } - - _cts.Dispose(); - } } } diff --git a/Tests/Xrpl.Tests/Client/PagedResponseServer.cs b/Tests/Xrpl.Tests/Client/PagedResponseServer.cs index 8162114c..ce23bd38 100644 --- a/Tests/Xrpl.Tests/Client/PagedResponseServer.cs +++ b/Tests/Xrpl.Tests/Client/PagedResponseServer.cs @@ -1,27 +1,20 @@ using System; using System.Buffers.Binary; -using System.IO; -using System.Net; using System.Net.Sockets; using System.Text; using System.Threading; using System.Threading.Tasks; -using Xrpl.Tests.MockRippled; - namespace Xrpl.Tests { /// - /// Minimal WebSocket server that answers every client request with a large, paged - /// rippled-shaped response echoing the request id. Each response is split into a controlled - /// number of WebSocket continuation frames, so the client assembles it from exactly that many - /// receive chunks — which is what a multi-megabyte ledger_data page looks like over a - /// real link. + /// WebSocket server that answers every client request with a large, paged rippled-shaped + /// response echoing the request id. Each response is split into a controlled number of + /// WebSocket continuation frames, so the client assembles it from exactly that many receive + /// chunks — which is what a multi-megabyte ledger_data page looks like over a real link. /// - internal sealed class PagedResponseServer : IDisposable + internal sealed class PagedResponseServer : WebSocketTestServerBase { - private readonly TcpListener _listener; - private readonly CancellationTokenSource _cts = new(); private readonly int _fragments; private readonly string _resultBody; private int _served; @@ -33,16 +26,9 @@ public PagedResponseServer(int approximatePayloadBytes, int fragments) _fragments = Math.Max(1, fragments); _resultBody = BuildResultBody(approximatePayloadBytes); - _listener = new TcpListener(IPAddress.Loopback, 0); - _listener.Start(); - Port = ((IPEndPoint)_listener.LocalEndpoint).Port; - _ = AcceptAsync(); + StartAccepting(); } - public int Port { get; } - - public string Url => "ws://127.0.0.1:" + Port + "/"; - /// Number of requests answered so far. public int Served => Volatile.Read(ref _served); @@ -91,40 +77,23 @@ private static void AppendHex(StringBuilder builder, int seed, int length) } } - private async Task AcceptAsync() + protected override async Task ServeAsync(NetworkStream stream) { - try + while (!Token.IsCancellationRequested) { - using TcpClient client = await _listener.AcceptTcpClientAsync(_cts.Token).ConfigureAwait(false); - client.NoDelay = true; - NetworkStream stream = client.GetStream(); - - string request = await ReadUntilHeadersEndAsync(stream).ConfigureAwait(false); - string key = Helpers.GetHandshakeRequestKey(request); - byte[] response = Encoding.ASCII.GetBytes(Helpers.GetHandshakeResponse(Helpers.HashKey(key))); - await stream.WriteAsync(response, _cts.Token).ConfigureAwait(false); - await stream.FlushAsync(_cts.Token).ConfigureAwait(false); - - while (!_cts.IsCancellationRequested) + string? message = await ReadTextFrameAsync(stream).ConfigureAwait(false); + if (message == null) { - string? message = await ReadTextFrameAsync(stream).ConfigureAwait(false); - if (message == null) - { - return; - } - - string id = ExtractId(message); - await WriteResponseAsync(stream, id).ConfigureAwait(false); - Interlocked.Increment(ref _served); + return; } - } - catch (OperationCanceledException) - { - // Normal shutdown. - } - catch (Exception ex) when (ex is IOException or SocketException or ObjectDisposedException) - { - // The client going away is the normal end of this loop. + + string id = ExtractId(message); + string envelope = "{\"id\":" + id + ",\"status\":\"success\",\"type\":\"response\",\"result\":" + + _resultBody + "}"; + + await WriteFragmentedMessageAsync(stream, Encoding.UTF8.GetBytes(envelope), _fragments) + .ConfigureAwait(false); + Interlocked.Increment(ref _served); } } @@ -164,59 +133,6 @@ private static string ExtractId(string message) return message.Substring(start, stop - start).Trim(); } - private async Task WriteResponseAsync(NetworkStream stream, string id) - { - string envelope = "{\"id\":" + id + ",\"status\":\"success\",\"type\":\"response\",\"result\":" + - _resultBody + "}"; - byte[] payload = Encoding.UTF8.GetBytes(envelope); - int fragmentBytes = (payload.Length + _fragments - 1) / _fragments; - - for (int fragment = 0; fragment < _fragments; fragment++) - { - // Ceil division can leave trailing frames past the end; they are sent empty - // so the frame count stays exactly as requested. - int offset = Math.Min(fragment * fragmentBytes, payload.Length); - int length = Math.Min(fragmentBytes, payload.Length - offset); - bool isFirst = fragment == 0; - bool isLast = fragment == _fragments - 1; - - await stream.WriteAsync(BuildFrameHeader(length, isFirst, isLast), _cts.Token) - .ConfigureAwait(false); - await stream.WriteAsync(payload.AsMemory(offset, length), _cts.Token).ConfigureAwait(false); - await stream.FlushAsync(_cts.Token).ConfigureAwait(false); - } - } - - /// - /// Unmasked server-to-client frame header. The first frame of a message carries the text - /// opcode (0x1), every following frame carries continuation (0x0); FIN is set on the last. - /// - private static byte[] BuildFrameHeader(int payloadLength, bool isFirst, bool isLast) - { - byte first = (byte)((isLast ? 0x80 : 0x00) | (isFirst ? 0x01 : 0x00)); - - if (payloadLength <= 125) - { - return new byte[] { first, (byte)payloadLength }; - } - - if (payloadLength <= ushort.MaxValue) - { - return new byte[] { first, 126, (byte)(payloadLength >> 8), (byte)payloadLength }; - } - - return new byte[] - { - first, - 127, - 0, 0, 0, 0, - (byte)(payloadLength >> 24), - (byte)(payloadLength >> 16), - (byte)(payloadLength >> 8), - (byte)payloadLength - }; - } - /// /// Reads one client frame. Returns the decoded text of the first text frame seen, or null /// once the peer closes. Control frames other than Close are skipped. @@ -287,71 +203,5 @@ private static byte[] BuildFrameHeader(int payloadLength, bool isFirst, bool isL } } } - - private async Task ReadExactAsync(NetworkStream stream, byte[] buffer, int count) - { - int read = 0; - - while (read < count) - { - int chunk = await stream.ReadAsync(buffer.AsMemory(read, count - read), _cts.Token) - .ConfigureAwait(false); - if (chunk == 0) - { - return false; - } - - read += chunk; - } - - return true; - } - - private async Task ReadUntilHeadersEndAsync(NetworkStream stream) - { - byte[] buffer = new byte[4096]; - StringBuilder request = new StringBuilder(); - - while (true) - { - int read = await stream.ReadAsync(buffer, _cts.Token).ConfigureAwait(false); - if (read == 0) - { - break; - } - - request.Append(Encoding.ASCII.GetString(buffer, 0, read)); - - if (request.ToString().Contains("\r\n\r\n", StringComparison.Ordinal)) - { - break; - } - } - - return request.ToString(); - } - - public void Dispose() - { - try - { - _cts.Cancel(); - } - catch - { - // best effort - } - - try - { - _listener.Stop(); - } - catch - { - // best effort - } - - _cts.Dispose(); - } } } diff --git a/Tests/Xrpl.Tests/Client/TestUWebSocketMessageAssembly.cs b/Tests/Xrpl.Tests/Client/TestUWebSocketMessageAssembly.cs index 4a4eee72..973388bc 100644 --- a/Tests/Xrpl.Tests/Client/TestUWebSocketMessageAssembly.cs +++ b/Tests/Xrpl.Tests/Client/TestUWebSocketMessageAssembly.cs @@ -124,7 +124,10 @@ private static async Task> ReceiveAsync(BulkMessageServer Task completed = await Task.WhenAny(done.Task, Task.Delay(TimeSpan.FromSeconds(WaitSeconds))) .ConfigureAwait(false); - Assert.AreSame(done.Task, completed, $"only {messages.Count} of {expected} messages arrived"); + Assert.AreSame( + done.Task, + completed, + $"only {messages.Count} of {expected} messages arrived; server fault: {server.Fault?.ToString() ?? "none"}"); } finally { diff --git a/Tests/Xrpl.Tests/Client/WebSocketTestServerBase.cs b/Tests/Xrpl.Tests/Client/WebSocketTestServerBase.cs new file mode 100644 index 00000000..9ac8d07a --- /dev/null +++ b/Tests/Xrpl.Tests/Client/WebSocketTestServerBase.cs @@ -0,0 +1,227 @@ +using System; +using System.Net; +using System.Net.Sockets; +using System.Text; +using System.Threading; +using System.Threading.Tasks; + +using Xrpl.Tests.MockRippled; + +namespace Xrpl.Tests +{ + /// + /// Shared plumbing for the raw WebSocket servers the client tests drive: a loopback listener, + /// the HTTP upgrade handshake, frame framing and disposal. The framing in particular lives here + /// on purpose — it used to be copied per server, and the copies drifted. + /// + internal abstract class WebSocketTestServerBase : IDisposable + { + private readonly TcpListener _listener; + private readonly CancellationTokenSource _cts = new CancellationTokenSource(); + private Exception? _fault; + + protected WebSocketTestServerBase() + { + // Captured once rather than read from the source on every use: Dispose does not wait for + // the accept loop, and reading CancellationTokenSource.Token after Dispose throws + // ObjectDisposedException, which would surface as a bogus Fault during teardown. Dispose + // always cancels before disposing, so a token captured here stays usable afterwards - + // an already cancelled token completes waits without touching its source. + Token = _cts.Token; + + _listener = new TcpListener(IPAddress.Loopback, 0); + _listener.Start(); + Port = ((IPEndPoint)_listener.LocalEndpoint).Port; + } + + /// Token cancelled when the server is disposed. + protected CancellationToken Token { get; } + + public int Port { get; } + + public string Url => "ws://127.0.0.1:" + Port + "/"; + + /// + /// Starts accepting. Call at the end of the derived constructor, once its own state is set: + /// the accept loop runs on the thread pool and may reach at any time. + /// + protected void StartAccepting() + { + _ = AcceptAsync(); + } + + /// Serves one connected client; the handshake has already completed. + protected abstract Task ServeAsync(NetworkStream stream); + + /// + /// The exception that ended the accept loop, if it ended badly. Recorded rather than + /// swallowed so a broken server shows up as itself instead of as the caller's timeout. + /// + public Exception? Fault => Volatile.Read(ref _fault); + + /// Called when the accept loop ends with an exception the server did not expect. + protected virtual void OnFaulted(Exception error) + { + Volatile.Write(ref _fault, error); + } + + /// Called when the accept loop ends because the server was disposed. + protected virtual void OnCancelled() + { + } + + private async Task AcceptAsync() + { + try + { + using TcpClient client = await _listener.AcceptTcpClientAsync(Token).ConfigureAwait(false); + client.NoDelay = true; + NetworkStream stream = client.GetStream(); + + string request = await ReadUntilHeadersEndAsync(stream).ConfigureAwait(false); + string key = Helpers.GetHandshakeRequestKey(request); + byte[] response = Encoding.ASCII.GetBytes(Helpers.GetHandshakeResponse(Helpers.HashKey(key))); + await stream.WriteAsync(response, Token).ConfigureAwait(false); + await stream.FlushAsync(Token).ConfigureAwait(false); + + await ServeAsync(stream).ConfigureAwait(false); + + // Hold the connection open until the test disposes the server. + await Task.Delay(Timeout.InfiniteTimeSpan, Token).ConfigureAwait(false); + } + catch (OperationCanceledException) + { + OnCancelled(); + } + catch (Exception ex) + { + OnFaulted(ex); + } + } + + /// + /// Unmasked server-to-client frame header. The first frame of a message carries the text + /// opcode (0x1), every following frame carries continuation (0x0); FIN is set on the last. + /// + protected static byte[] BuildFrameHeader(int payloadLength, bool isFirst, bool isLast) + { + byte first = (byte)((isLast ? 0x80 : 0x00) | (isFirst ? 0x01 : 0x00)); + + if (payloadLength <= 125) + { + return new byte[] { first, (byte)payloadLength }; + } + + if (payloadLength <= ushort.MaxValue) + { + return new byte[] { first, 126, (byte)(payloadLength >> 8), (byte)payloadLength }; + } + + return new byte[] + { + first, + 127, + 0, 0, 0, 0, + (byte)(payloadLength >> 24), + (byte)(payloadLength >> 16), + (byte)(payloadLength >> 8), + (byte)payloadLength + }; + } + + /// + /// Writes one message split into frames. Ceil division can + /// leave trailing frames past the end of the payload; they are sent empty so the frame + /// count is exactly what the caller asked for. + /// + protected async Task WriteFragmentedMessageAsync(NetworkStream stream, ReadOnlyMemory payload, int fragments) + { + int fragmentBytes = (payload.Length + fragments - 1) / fragments; + + for (int fragment = 0; fragment < fragments; fragment++) + { + int offset = Math.Min(fragment * fragmentBytes, payload.Length); + int length = Math.Min(fragmentBytes, payload.Length - offset); + bool isFirst = fragment == 0; + bool isLast = fragment == fragments - 1; + + await stream.WriteAsync(BuildFrameHeader(length, isFirst, isLast), Token).ConfigureAwait(false); + await stream.WriteAsync(payload.Slice(offset, length), Token).ConfigureAwait(false); + await stream.FlushAsync(Token).ConfigureAwait(false); + } + } + + protected async Task ReadExactAsync(NetworkStream stream, byte[] buffer, int count) + { + int read = 0; + + while (read < count) + { + int chunk = await stream.ReadAsync(buffer.AsMemory(read, count - read), Token).ConfigureAwait(false); + if (chunk == 0) + { + return false; + } + + read += chunk; + } + + return true; + } + + /// + /// Reads the client's HTTP upgrade request up to the blank line that ends the headers. + /// Anything the same read happened to pull in after that terminator is returned as part of + /// the string and then dropped by the caller: no client here pipelines a WebSocket frame + /// onto the upgrade request, because it has to wait for the 101 before it may send one. + /// A server that ever needs to accept such a client has to hand the leftover bytes to + /// instead. + /// + private async Task ReadUntilHeadersEndAsync(NetworkStream stream) + { + byte[] buffer = new byte[4096]; + StringBuilder request = new StringBuilder(); + + while (true) + { + int read = await stream.ReadAsync(buffer, Token).ConfigureAwait(false); + if (read == 0) + { + break; + } + + request.Append(Encoding.ASCII.GetString(buffer, 0, read)); + + if (request.ToString().Contains("\r\n\r\n", StringComparison.Ordinal)) + { + break; + } + } + + return request.ToString(); + } + + public void Dispose() + { + try + { + _cts.Cancel(); + } + catch + { + // best effort + } + + try + { + _listener.Stop(); + } + catch + { + // best effort + } + + _cts.Dispose(); + } + } +} diff --git a/Xrpl/Client/WebSocketClient.cs b/Xrpl/Client/WebSocketClient.cs index 8a0c885d..e4ae5693 100644 --- a/Xrpl/Client/WebSocketClient.cs +++ b/Xrpl/Client/WebSocketClient.cs @@ -484,7 +484,9 @@ private async Task ReceiveLoopAsync() // Scratch buffer for messages that arrive in more than one chunk. It grows to the // largest message seen on this connection and is then reused, so steady-state assembly - // costs nothing beyond the single exact-sized array handed to the callbacks. + // costs nothing beyond the single exact-sized array handed to the callbacks. Rented for + // the same reason as the receive buffer above - at these sizes it is a large-object-heap + // array that would otherwise be thrown away per connection. byte[]? assemblyBuffer = null; try @@ -637,6 +639,11 @@ private async Task ReceiveLoopAsync() finally { ArrayPool.Shared.Return(buffer); + + if (assemblyBuffer != null) + { + ArrayPool.Shared.Return(assemblyBuffer); + } } } @@ -658,12 +665,19 @@ private static void EnsureAssemblyCapacity(ref byte[]? assemblyBuffer, int requi capacity = capacity <= Array.MaxLength / 2 ? capacity * 2 : requiredLength; } - byte[] grown = new byte[capacity]; + // Rent may hand back a longer array than asked for; the growth above keys off the + // actual length, so the next doubling starts from what we really got. + byte[] grown = ArrayPool.Shared.Rent(capacity); if (preserveLength > 0) { Buffer.BlockCopy(assemblyBuffer!, 0, grown, 0, preserveLength); } + if (assemblyBuffer != null) + { + ArrayPool.Shared.Return(assemblyBuffer); + } + assemblyBuffer = grown; }