From d530b7021288b8c9d9fec4610477df1d71e15f7e Mon Sep 17 00:00:00 2001 From: hchen Date: Tue, 15 Sep 2026 11:51:30 -0500 Subject: [PATCH] Introduce async ChatEventDispatcher for non-blocking events Added ChatEventDispatcher to decouple chat event production from SignalR delivery, ensuring producers are never blocked by slow clients. Registered dispatcher as singleton with detailed documentation. Refactored event emission to use dispatcher queue, updated observer to avoid awaiting sends. Added unit tests for dispatcher behavior and updated test project references. --- .../BotSharp.Plugin.ChatHub/ChatHubPlugin.cs | 7 +- .../Helpers/ChatEventDispatcher.cs | 161 +++++++++++++++++ .../Helpers/EventEmitter.cs | 44 ++++- .../Observers/ChatHubObserver.cs | 10 +- .../BotSharp.Core.UnitTests.csproj | 2 + .../ChatHub/ChatEventDispatcherTests.cs | 166 ++++++++++++++++++ 6 files changed, 383 insertions(+), 7 deletions(-) create mode 100644 src/Plugins/BotSharp.Plugin.ChatHub/Helpers/ChatEventDispatcher.cs create mode 100644 tests/BotSharp.Core.UnitTests/ChatHub/ChatEventDispatcherTests.cs diff --git a/src/Plugins/BotSharp.Plugin.ChatHub/ChatHubPlugin.cs b/src/Plugins/BotSharp.Plugin.ChatHub/ChatHubPlugin.cs index 4ba2e30c0..9708612e9 100644 --- a/src/Plugins/BotSharp.Plugin.ChatHub/ChatHubPlugin.cs +++ b/src/Plugins/BotSharp.Plugin.ChatHub/ChatHubPlugin.cs @@ -1,4 +1,4 @@ -using BotSharp.Abstraction.MessageHub.Models; +using BotSharp.Abstraction.MessageHub.Models; using BotSharp.Abstraction.MessageHub.Observers; using BotSharp.Plugin.ChatHub.Hooks; using BotSharp.Plugin.ChatHub.Observers; @@ -22,6 +22,11 @@ public void RegisterDI(IServiceCollection services, IConfiguration config) config.Bind("ChatHub", settings); services.AddSingleton(x => settings); + // One per process: it holds a delivery chain per target, so it has to outlive any scope. + // Everything that reaches SignalR goes through it, which is what keeps a stalled client + // from stopping the thread that produced the event. See ChatEventDispatcher. + services.AddSingleton(); + services.AddScoped>, ChatHubObserver>(); // Register hooks diff --git a/src/Plugins/BotSharp.Plugin.ChatHub/Helpers/ChatEventDispatcher.cs b/src/Plugins/BotSharp.Plugin.ChatHub/Helpers/ChatEventDispatcher.cs new file mode 100644 index 000000000..c9be1c6a3 --- /dev/null +++ b/src/Plugins/BotSharp.Plugin.ChatHub/Helpers/ChatEventDispatcher.cs @@ -0,0 +1,161 @@ +using System.Collections.Concurrent; +using System.Threading; + +namespace BotSharp.Plugin.ChatHub.Helpers; + +/// +/// Hands chat events to SignalR without ever making the caller wait for the network. +/// +/// +/// +/// WHY THIS EXISTS. Sending an event used to be a plain await on +/// IClientProxy.SendAsync, and reached it through +/// GetAwaiter().GetResult() — a synchronous, unbounded, uncancellable wait on a socket. +/// That observer runs on whatever thread pushed to the MessageHub, and for a tool's +/// notifications/progress that thread is the MCP client's own read loop (McpToolExecutor +/// supplies the progress reporter, and the MCP SDK invokes it inline on the loop). So one client +/// that had stopped reading — a dead browser tab, a socket a proxy had dropped without a FIN — +/// stopped the read loop, and the tool call whose progress was being relayed never saw its own +/// response: it neither returned nor threw, not even at the HttpClient timeout, because that +/// continuation needs the loop that is stalled. A ten-minute browser task finished perfectly and +/// the agent's turn was already dead (OneFlow, 2026-09-15). +/// +/// +/// Since Subject.Synchronize serialises every push, one such send also held up chat events +/// for every other conversation in the process. +/// +/// +/// WHAT THIS DOES. is O(1) and returns at once; delivery happens on the +/// thread pool. Three properties matter and each answers a way the naive fix goes wrong: +/// +/// +/// ORDER IS KEPT, per target. Indications are a narration — "Now on plan step 2.1", then +/// "Clicking SEND" — and the one that arrives last is the one left on screen, so delivering them +/// concurrently would show them out of order. Each target has its own chain and nothing overtakes +/// within it. Two targets never wait for each other, which is the property the old code lacked. +/// A WEDGED SEND CANNOT OWN ITS CHAIN FOREVER. Each delivery carries +/// , so a socket that will never drain costs that conversation one +/// timeout rather than everything queued behind it for the life of the process. Nothing blocks +/// while that timer runs — it only bounds a task. +/// THE QUEUE IS BOUNDED. A client that stops reading entirely would otherwise accumulate one +/// entry per event for as long as the conversation lives. Past the event +/// is dropped and said so in the log; by then that browser is not showing anything anyway, and the +/// conversation itself is persisted elsewhere — SignalR is the live view, not the record. +/// +/// +/// Nothing here may throw into a caller: a send that fails is logged and the chain moves on. The +/// lanes clean themselves up — the last delivery for a target removes it — so an idle process +/// holds no entry per conversation it has ever seen. +/// +/// +public sealed class ChatEventDispatcher +{ + /// How long one delivery may take before it is abandoned. See the remarks. + public static readonly TimeSpan DefaultSendTimeout = TimeSpan.FromSeconds(30); + + /// How many events may be waiting for one target before further ones are dropped. + public const int DefaultMaxQueued = 256; + + private readonly ConcurrentDictionary _lanes = new(); + private readonly TimeSpan _sendTimeout; + private readonly int _maxQueued; + + public ChatEventDispatcher() : this(DefaultSendTimeout, DefaultMaxQueued) + { + } + + /// Overridable for tests, which cannot wait out the real timeout. + public ChatEventDispatcher(TimeSpan sendTimeout, int maxQueued) + { + _sendTimeout = sendTimeout; + _maxQueued = maxQueued; + } + + /// + /// Queue one delivery for and return. Never blocks, never throws. + /// runs on the thread pool, after everything already queued for the + /// same target and never concurrently with it. + /// + /// + /// What the event is addressed to — the group or the user, prefixed with which — so the two + /// dispatch modes cannot share a chain. + /// + /// Named in the log if the delivery fails or is dropped. + public void Enqueue(string target, Func send, ILogger logger, string description) + { + while (true) + { + var lane = _lanes.GetOrAdd(target, _ => new Lane()); + + lock (lane.Gate) + { + // The lane retired between GetOrAdd and this lock — its last delivery finished. + // Drop it (a no-op if the retiring side already did) and take a fresh one. + if (lane.Retired) + { + _lanes.TryRemove(new KeyValuePair(target, lane)); + continue; + } + + if (lane.Pending >= _maxQueued) + { + logger.LogWarning( + $"Dropped chat event '{description}' for {target}: {lane.Pending} already waiting, so the client is not reading."); + return; + } + + lane.Pending++; + lane.Tail = Deliver(lane.Tail, target, lane, send, logger, description); + } + + return; + } + } + + /// + /// The number of targets currently holding a queue. For tests and diagnostics: a healthy + /// process settles back to zero. + /// + public int ActiveTargets => _lanes.Count; + + private Task Deliver(Task tail, string target, Lane lane, Func send, ILogger logger, string description) + { + // TaskScheduler.Default, and no ExecuteSynchronously: a tail that is already complete must + // still be continued on the thread pool. Inlining it here would run the send on the caller + // — which is the thread this class exists to keep free. + return tail.ContinueWith(async _ => + { + try + { + using var cts = new CancellationTokenSource(_sendTimeout); + await send(cts.Token); + } + catch (Exception ex) + { + logger.LogWarning(ex, $"Failed to send chat event '{description}' to {target}"); + } + finally + { + lock (lane.Gate) + { + if (--lane.Pending == 0) + { + // Nothing left for this target. Retire under the same lock an enqueue takes, + // so a writer either gets in before this and keeps the lane alive, or sees + // Retired and starts a new one. It cannot land on a lane nothing will drain. + lane.Retired = true; + _lanes.TryRemove(new KeyValuePair(target, lane)); + } + } + } + }, CancellationToken.None, TaskContinuationOptions.DenyChildAttach, TaskScheduler.Default).Unwrap(); + } + + private sealed class Lane + { + public readonly object Gate = new(); + public Task Tail = Task.CompletedTask; + public int Pending; + public bool Retired; + } +} diff --git a/src/Plugins/BotSharp.Plugin.ChatHub/Helpers/EventEmitter.cs b/src/Plugins/BotSharp.Plugin.ChatHub/Helpers/EventEmitter.cs index 821b173e3..3686cbb7e 100644 --- a/src/Plugins/BotSharp.Plugin.ChatHub/Helpers/EventEmitter.cs +++ b/src/Plugins/BotSharp.Plugin.ChatHub/Helpers/EventEmitter.cs @@ -1,11 +1,29 @@ using Microsoft.AspNetCore.SignalR; using System.Runtime.CompilerServices; +using System.Threading; namespace BotSharp.Plugin.ChatHub.Helpers; internal class EventEmitter { - internal static async Task SendChatEvent( + /// + /// Address one chat event and hand it to . + /// + /// + /// + /// Returns as soon as the event is QUEUED, not when the client has it. The awaits at the call + /// sites are therefore free, and the one caller that cannot await — ChatHubObserver, + /// which is an IObserver and runs on whatever thread pushed to the MessageHub — no + /// longer stops that thread on a socket. See the dispatcher for what a stalled client used to + /// cost. + /// + /// + /// Everything that needs the request's services is resolved HERE, on the caller's thread, and + /// captured: the proxy is taken from the singleton hub context, so a delivery running after the + /// scope has been disposed still has what it needs. + /// + /// + internal static Task SendChatEvent( IServiceProvider services, ILogger logger, string @event, @@ -20,20 +38,40 @@ internal static async Task SendChatEvent( { var settings = services.GetRequiredService(); var chatHub = services.GetRequiredService>(); + var dispatcher = services.GetRequiredService(); + + IClientProxy? clients = null; + string? target = null; switch (settings.EventDispatchBy) { case EventDispatchType.Group when !string.IsNullOrEmpty(conversationId): - await chatHub.Clients.Group(conversationId).SendAsync(@event, data); + clients = chatHub.Clients.Group(conversationId); + // Prefixed so the two dispatch modes cannot share one delivery chain. + target = $"{EventDispatchType.Group}:{conversationId}"; break; case EventDispatchType.User when !string.IsNullOrEmpty(userId): - await chatHub.Clients.User(userId).SendAsync(@event, data); + clients = chatHub.Clients.User(userId); + target = $"{EventDispatchType.User}:{userId}"; break; } + + if (clients == null || target == null) + { + return Task.CompletedTask; + } + + dispatcher.Enqueue( + target, + ct => clients.SendAsync(@event, data, ct), + logger, + $"{@event} ({callerClass}-{callerMethod}) (conversation id: {conversationId})"); } catch (Exception ex) { logger.Log(logLevel, ex, $"Failed to send event '{@event}' in ({callerClass}-{callerMethod}) (conversation id: {conversationId})"); } + + return Task.CompletedTask; } } diff --git a/src/Plugins/BotSharp.Plugin.ChatHub/Observers/ChatHubObserver.cs b/src/Plugins/BotSharp.Plugin.ChatHub/Observers/ChatHubObserver.cs index 614340cfd..f15d27a8b 100644 --- a/src/Plugins/BotSharp.Plugin.ChatHub/Observers/ChatHubObserver.cs +++ b/src/Plugins/BotSharp.Plugin.ChatHub/Observers/ChatHubObserver.cs @@ -1,4 +1,4 @@ -using BotSharp.Abstraction.Conversations.Dtos; +using BotSharp.Abstraction.Conversations.Dtos; using BotSharp.Abstraction.Conversations.Enums; using BotSharp.Abstraction.MessageHub.Models; using BotSharp.Abstraction.MessageHub.Observers; @@ -159,8 +159,12 @@ private void SendEvent(string @event, string conversationId, T data, [CallerM { var user = _services.GetRequiredService(); var json = JsonSerializer.Serialize(data, _options.JsonSerializerOptions); - EventEmitter.SendChatEvent(_services, _logger, @event, conversationId, user?.Id, json, nameof(ChatHubObserver), callerName) - .ConfigureAwait(false).GetAwaiter().GetResult(); + + // NOT awaited, and it does not need to be: SendChatEvent queues the event and returns. + // This method is called from OnNext, on whatever thread pushed to the MessageHub — for a + // tool's progress notifications, the MCP client's own read loop — so anything that waited + // here for a socket could stop that thread, and did. See ChatEventDispatcher. + _ = EventEmitter.SendChatEvent(_services, _logger, @event, conversationId, user?.Id, json, nameof(ChatHubObserver), callerName); } #endregion } diff --git a/tests/BotSharp.Core.UnitTests/BotSharp.Core.UnitTests.csproj b/tests/BotSharp.Core.UnitTests/BotSharp.Core.UnitTests.csproj index 030ffeaa7..a56498e7e 100644 --- a/tests/BotSharp.Core.UnitTests/BotSharp.Core.UnitTests.csproj +++ b/tests/BotSharp.Core.UnitTests/BotSharp.Core.UnitTests.csproj @@ -32,6 +32,8 @@ + + diff --git a/tests/BotSharp.Core.UnitTests/ChatHub/ChatEventDispatcherTests.cs b/tests/BotSharp.Core.UnitTests/ChatHub/ChatEventDispatcherTests.cs new file mode 100644 index 000000000..b72eef5eb --- /dev/null +++ b/tests/BotSharp.Core.UnitTests/ChatHub/ChatEventDispatcherTests.cs @@ -0,0 +1,166 @@ +using System.Diagnostics; +using BotSharp.Plugin.ChatHub.Helpers; +using Microsoft.Extensions.Logging.Abstractions; +using Xunit; + +namespace BotSharp.Core.UnitTests.ChatHub; + +/// +/// Chat events must never make their producer wait for a client. +/// +/// The producer is often not a request thread: ChatHubObserver runs on whatever pushed to the +/// MessageHub, and for a tool's progress notifications that is the MCP client's own read loop. A +/// send that waited there stopped the loop, so the tool call being narrated never saw its own +/// response — it neither returned nor threw. These pin the properties that close that off. +/// +public class ChatEventDispatcherTests +{ + private static readonly TimeSpan ShortTimeout = TimeSpan.FromMilliseconds(300); + + private static ChatEventDispatcher Dispatcher(int maxQueued = 256) + => new(ShortTimeout, maxQueued); + + private static void Enqueue(ChatEventDispatcher dispatcher, string target, Func send) + => dispatcher.Enqueue(target, send, NullLogger.Instance, "test event"); + + [Fact] + public async Task EnqueueDoesNotWaitForTheSend() + { + var dispatcher = Dispatcher(); + var stuck = new TaskCompletionSource(); + var entered = new TaskCompletionSource(); + + var watch = Stopwatch.StartNew(); + Enqueue(dispatcher, "group:a", _ => + { + entered.TrySetResult(); + return stuck.Task; + }); + watch.Stop(); + + Assert.True(watch.ElapsedMilliseconds < 200, $"Enqueue took {watch.ElapsedMilliseconds}ms — it waited for the send."); + + // And the send really did start, so the call above returned while it was in flight rather + // than because nothing happened. + await entered.Task.WaitAsync(TimeSpan.FromSeconds(5)); + stuck.TrySetResult(); + } + + [Fact] + public async Task OneStalledTargetDoesNotHoldUpAnother() + { + var dispatcher = Dispatcher(); + var stuck = new TaskCompletionSource(); + var delivered = new TaskCompletionSource(); + + Enqueue(dispatcher, "group:stalled", _ => stuck.Task); + Enqueue(dispatcher, "group:healthy", _ => + { + delivered.TrySetResult(); + return Task.CompletedTask; + }); + + await delivered.Task.WaitAsync(TimeSpan.FromSeconds(5)); + stuck.TrySetResult(); + } + + [Fact] + public async Task DeliveriesToOneTargetKeepTheirOrder() + { + var dispatcher = Dispatcher(); + var order = new List(); + var all = new TaskCompletionSource(); + const int count = 50; + + for (var i = 0; i < count; i++) + { + var n = i; + Enqueue(dispatcher, "group:a", async _ => + { + // Uneven work: concurrent delivery would reorder these, a serial chain cannot. + await Task.Delay(n % 3); + order.Add(n); + if (order.Count == count) all.TrySetResult(); + }); + } + + await all.Task.WaitAsync(TimeSpan.FromSeconds(30)); + Assert.Equal(Enumerable.Range(0, count), order); + } + + [Fact] + public async Task AWedgedSendIsAbandonedSoTheNextOneRuns() + { + var dispatcher = Dispatcher(); + var cancelled = new TaskCompletionSource(); + var next = new TaskCompletionSource(); + + // Never completes on its own; only the dispatcher's own timeout can end it. + Enqueue(dispatcher, "group:a", async ct => + { + try + { + await Task.Delay(Timeout.Infinite, ct); + } + catch (OperationCanceledException) + { + cancelled.TrySetResult(); + throw; + } + }); + Enqueue(dispatcher, "group:a", _ => + { + next.TrySetResult(); + return Task.CompletedTask; + }); + + await cancelled.Task.WaitAsync(TimeSpan.FromSeconds(10)); + await next.Task.WaitAsync(TimeSpan.FromSeconds(10)); + } + + [Fact] + public async Task AClientThatNeverReadsCannotQueueWithoutBound() + { + var dispatcher = Dispatcher(maxQueued: 4); + var stuck = new TaskCompletionSource(); + var started = 0; + + for (var i = 0; i < 100; i++) + { + Enqueue(dispatcher, "group:a", _ => + { + Interlocked.Increment(ref started); + return stuck.Task; + }); + } + + stuck.TrySetResult(); + await WaitUntil(() => dispatcher.ActiveTargets == 0, TimeSpan.FromSeconds(10)); + + // The first is in flight and at most `maxQueued` are ever waiting, so nothing near 100 + // was accepted. The rest were dropped with a warning rather than held. + Assert.InRange(Volatile.Read(ref started), 1, 8); + } + + [Fact] + public async Task AnIdleDispatcherHoldsNothing() + { + var dispatcher = Dispatcher(); + for (var i = 0; i < 20; i++) + { + Enqueue(dispatcher, $"group:{i}", _ => Task.CompletedTask); + } + + await WaitUntil(() => dispatcher.ActiveTargets == 0, TimeSpan.FromSeconds(10)); + Assert.Equal(0, dispatcher.ActiveTargets); + } + + private static async Task WaitUntil(Func condition, TimeSpan limit) + { + var watch = Stopwatch.StartNew(); + while (!condition() && watch.Elapsed < limit) + { + await Task.Delay(10); + } + } +}