Обработчик потока выполняется внутри приёмного цикла: медленный подписчик останавливает всё соединение

[Platonenkov]

Наблюдение

Приёмный цикл ожидает завершения пользовательского обработчика прямо внутри себя — Xrpl/Client/connection.cs, диспетчеризация потоков:

// строка ~3049
if (OnLedgerClosed is not null)
    await OnLedgerClosed.Invoke(response)!;

// строка ~3069
if (OnTransaction is not null)
    await OnTransaction.Invoke(response)!;

То же и для OnValidationReceived рядом.

Почему это проблема

Обработчик подписчика становится частью критического пути соединения. Пока он выполняется, цикл не читает следующее сообщение — а по этому же соединению идут обычные запросы-ответы.

Практические следствия:

• обработчик, делающий что-либо небыстрое (запись в БД, сетевой вызов, ожидание блокировки), задерживает не только поток, но и все параллельные запросы на том же клиенте;
• при активной подписке на transactions сообщения приходят непрерывно, поэтому даже небольшая задержка на сообщение накапливается и рано или поздно упирается в таймаут запроса;
• обработчик, который завис или упал в дедлок, полностью останавливает соединение, при этом внешне оно выглядит живым.

Отдельная сторона того же — отсутствие определённой семантики при отставании подписчика. Сейчас поведение «медленный потребитель» не описано ни в контракте, ни в доках: неизвестно, ждёт ли SDK, теряет ли сообщения, растёт ли где-то очередь.

Ожидаемое

Доставка сообщений потока развязана с приёмным циклом, и поведение при отставании подписчика определено явно. Разумные варианты (выбор за реализацией):

• очередь с ограничением (System.Threading.Channels) и заданной политикой переполнения — ждать, отбрасывать старые или отбрасывать новые, но выбор задокументирован и по возможности настраиваем;
• либо явно оставить синхронную доставку, но задокументировать её как контракт: «обработчик обязан возвращать управление немедленно, вся работа — за его пределами».

Второе дешевле и тоже приемлемо. Неприемлемо нынешнее состояние, где семантика не описана, а цена медленного обработчика — остановка всего соединения.

Связь с другими issue

Если контракт подписки будет выведен на IXrplClient (#103), выбранная там форма доставки фактически решает и этот вопрос — их разумно рассматривать вместе. Но дефект существует и сегодня, независимо от того, через что подписываться.

Проверка

Тест с намеренно медленным обработчиком: обычные запросы на том же соединении продолжают отвечать в разумное время, поведение потока соответствует задокументированному.

[Platonenkov]

Проверил по коду — описанной механики нет, и не было на момент заведения issue. Но на этом же месте есть другая проблема, которая, кажется, и стоит за наблюдением.

Обработчик не выполняется в приёмном цикле

Строки с await OnTransaction.Invoke(response) указаны верно, но они лежат внутри ProcessStreamMessageAsync, а его вызывает не приёмный цикл, а отдельная задача, читающая канал (connection.cs, StartMessageProcessor):

_messageProcessorTask = Task.Run(async () =>
{
    var reader = channel.Reader;
    while (await reader.WaitToReadAsync(cts.Token).ConfigureAwait(false))
    {
        while (reader.TryRead(out var frame))
        {
            await ProcessStreamMessageAsync(frame).ConfigureAwait(false);
        }
    }
});

Приёмный цикл до обработчика не доходит вовсе — он кладёт кадр неблокирующим TryWrite и идёт читать следующее сообщение:

private void EnqueueStreamMessage(byte[] frame)
{
    ...
    channel.Writer.TryWrite(frame);
}

StartMessageProcessor() вызывается при каждом успешном подключении (connection.cs:1872), так что путь через канал — основной, а не какая-то опция.

Канал появился коммитом 3c7e38e ещё в феврале 2026 — то есть за полгода до заведения этого issue. Так что медленный обработчик не задерживает ни поток, ни параллельные запросы: и то и другое читается приёмным циклом, который в это время свободен.

Что на этом месте есть на самом деле

Канал ограничен и настроен на вытеснение:

Channel.CreateBounded<byte[]>(new BoundedChannelOptions(10000)
{
    SingleReader = true,
    SingleWriter = false,
    FullMode = BoundedChannelFullMode.DropOldest
});

То есть медленный подписчик приводит не к остановке соединения, а к молчаливой потере событий: при отставании на 10 000 сообщений начинают выбрасываться самые старые. Ни исключения, ни лога — при DropOldest TryWrite всегда возвращает true, поэтому ветки «канал полон» здесь не может быть в принципе (я убрал соответствующую недостижимую строку лога в 127ebca, оставив вместо неё комментарий о самой потере).

Для потребителя, строящего состояние по потоку, это тот же класс отказа, что описан в #104: расхождение с леджером без единого признака, что что-то пропало.

Отдельно: под WebAssembly и в случае, когда канал ещё не создан, используется путь _ = ProcessStreamMessageFireAndForgetAsync(frame). Он тоже не блокирует приёмный цикл, но у него нет ни ограничения, ни упорядоченности — при быстром потоке обработчики поедут параллельно и порядок событий не гарантирован, в отличие от SingleReader = true на основном пути.

Предложение

Переформулировать issue с «медленный подписчик останавливает соединение» на «отставание подписчика теряет события молча», а решением считать наблюдаемость и выбор политики:

1. сообщать о вытеснении — счётчик потерянных кадров, событие или лог, чтобы потеря перестала быть невидимой;
2. дать выбор политики вместо жёсткого DropOldest — например Wait с явным ограничением для потребителей, которым важна полнота, а не свежесть;
3. отдельно описать поведение под WebAssembly, где порядок не гарантирован.

Если переформулировка не нужна — issue стоит закрыть, а пункты выше завести отдельно: чинить в текущей формулировке нечего.

[Platonenkov]

Закрываю: описанной механики нет, а то реальное, что стояло за наблюдением, закрыто.

Разбор в комментарии выше от 19 августа: строки с await OnTransaction.Invoke(response) указаны верно, но лежат внутри ProcessStreamMessageAsync, а его вызывает не приёмный цикл, а отдельная задача-читатель канала. Медленный подписчик соединение не останавливает.

Что за этим стояло на самом деле — молчаливая потеря событий — закрыто в двух шагах:

• Потерянные события потока перестают быть невидимыми: счётчик вытеснения и размер очереди #109: DroppedStreamMessages и настраиваемый StreamMessageQueueCapacity. Очередь вытесняла старейшее событие без единого следа, так что «подписчик отстал» и «в сети ничего не происходит» выглядели одинаково;
• Один путь потока для всех платформ: браузер перестаёт обходить очередь #111: под WebAssembly очереди на пути не было вовсе, то есть наблюдаемость из Потерянные события потока перестают быть невидимыми: счётчик вытеснения и размер очереди #109 отсутствовала ровно там, где что-либо наблюдать труднее всего. Теперь путь один для всех платформ.

Попутно в #111 добавлен FallbackDispatchedStreamMessages — он считает кадры, выданные в обход очереди, и делает измеримым остаток, вынесенный в #113.