From 07b4b2ae32eea0cf37d479c1685ca3744b1fe9bf Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Roland=20B=C3=BCrgi?= Date: Mon, 6 Jul 2026 23:15:03 +0200 Subject: [PATCH 1/2] =?UTF-8?q?fix(messaging):=20snapshot=20the=20rule=20c?= =?UTF-8?q?hain=20=E2=80=94=20concurrent=20Remove=20NREs=20the=20dispatch?= =?UTF-8?q?=20walk?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit MessageHub dispatches a delivery by folding over its rule chain (`rules`, a ThreadSafeLinkedList). HandleMessageAsync walked the chain via the raw LinkedListNode.Next OUTSIDE the list's lock. ThreadSafeLinkedList locks Add/Remove/First but not that walk, so when a handler-disposable's rules.Remove(node) fires concurrently (rapid sync-hub create/teardown churn under load), LinkedList.Remove→Invalidate nulls the removed node's owning-list reference before its next pointer, and a racing get_Next() dereferences list.head → NullReferenceException. That fails the delivery → DeliveryFailure → the synchronization stream reports [SYNC_STREAM] OnError and propagates a StreamErrorEvent → the client subscriber's .Within(Ns) never matches → timeout. Because it depends on which sync hub is being torn down at that instant, a different sync-hub layout test flakes each bulk run (EditorTest.TestEditorWithDelayed, EditPersistenceTest.EditState_ShouldSurviveDataUpdates, …). Fix: iterate a snapshot taken under the read lock instead of walking live nodes. - ThreadSafeLinkedList.Snapshot() copies the values under the read lock. - MessageHub.HandleMessageAsync folds over the snapshot array; a concurrent Remove can no longer invalidate the iteration. Also semantically correct: a delivery is handled by the rules present when dispatch began. Repro: `dotnet test test/MeshWeaver.Layout.Test` in a loop — ~1-in-5 runs failed with the NRE before, 9+ clean after. Co-Authored-By: Claude Opus 4.8 (1M context) --- src/MeshWeaver.Messaging.Hub/MessageHub.cs | 32 +++++++++++-------- .../ThreadSafeLinkedList.cs | 24 ++++++++++++++ 2 files changed, 42 insertions(+), 14 deletions(-) diff --git a/src/MeshWeaver.Messaging.Hub/MessageHub.cs b/src/MeshWeaver.Messaging.Hub/MessageHub.cs index 58241f863e..776446a29d 100644 --- a/src/MeshWeaver.Messaging.Hub/MessageHub.cs +++ b/src/MeshWeaver.Messaging.Hub/MessageHub.cs @@ -675,7 +675,7 @@ private record TypeAndHandler(Type Type, AsyncDelivery? Action); private IObservable HandleMessageAsync( IMessageDelivery delivery, - LinkedListNode node, + AsyncDelivery[] ruleChain, CancellationToken cancellationToken ) { @@ -686,18 +686,19 @@ CancellationToken cancellationToken // subscribe (on the action-block thread), genuinely-async rules complete // later via the pool. The chain is BUILT here (one SelectMany per rule); // it runs when the actor-loop edge subscribes (.ToTask). - var inputMessageType = delivery.Message.GetType(); + // + // 🚨 ruleChain is a SNAPSHOT taken under ThreadSafeLinkedList's read lock — NOT a live + // walk of LinkedListNode.Next. A raw .Next walk races a concurrent rules.Remove(node) + // (a handler disposable firing during hub teardown / rapid sync-hub churn): LinkedList + // invalidates the removed node's owning-list reference before its next pointer, so a + // racing get_Next() dereferences list.head and throws NRE → the delivery fails → sync + // streams see it as [SYNC_STREAM] OnError and their subscribers time out (a different + // sync-hub test flakes each bulk run). Iterating the snapshot is immune to that race. + if (ruleChain.Length > 500) + throw new InvalidOperationException($"HandleMessageAsync rule count exceeded 500 in hub {Address} for {delivery.Message.GetType().Name}"); IObservable result = Observable.Return(delivery); - LinkedListNode? current = node; - var depth = 0; - while (current is not null) - { - if (depth++ > 500) - throw new InvalidOperationException($"HandleMessageAsync recursion depth exceeded 500 in hub {Address} for {inputMessageType.Name}"); - var rule = current.Value; + foreach (var rule in ruleChain) result = result.SelectMany(d => rule.Invoke(d, cancellationToken)); - current = current.Next; - } return result; } @@ -752,14 +753,17 @@ CancellationToken cancellationToken && !AccessService.LooksLikeHubPrincipal(delivery.AccessContext.ObjectId)) accessService.SetContext(delivery.AccessContext); + // Snapshot the rule chain ONCE under the list's read lock (see HandleMessageAsync) so a + // concurrent rules.Remove during teardown can't NRE the iteration. + var ruleChain = rules.Snapshot(); if (traceEnabled) - logger.LogTrace(rules.First != null + logger.LogTrace(ruleChain.Length > 0 ? "MESSAGE_FLOW: HUB_PROCESSING_RULES | {MessageType} | Hub: {Address} | MessageId: {MessageId}" : "MESSAGE_FLOW: HUB_NO_RULES | {MessageType} | Hub: {Address} | MessageId: {MessageId}", messageTypeName, Address, delivery.Id); - var chain = rules.First != null - ? HandleMessageAsync(delivery, rules.First, cancellationToken) + var chain = ruleChain.Length > 0 + ? HandleMessageAsync(delivery, ruleChain, cancellationToken) : Observable.Return(delivery); return chain diff --git a/src/MeshWeaver.Messaging.Hub/ThreadSafeLinkedList.cs b/src/MeshWeaver.Messaging.Hub/ThreadSafeLinkedList.cs index 9a144094b1..93070c0a60 100644 --- a/src/MeshWeaver.Messaging.Hub/ThreadSafeLinkedList.cs +++ b/src/MeshWeaver.Messaging.Hub/ThreadSafeLinkedList.cs @@ -62,6 +62,30 @@ public void Remove(LinkedListNode item) } } + /// + /// Returns a point-in-time copy of the values (head→tail) taken under the read lock. + /// Callers that need to ITERATE the list MUST use this instead of walking + /// off : a raw .Next walk runs + /// outside the lock, and a concurrent invalidates the node mid-walk + /// ( nulls the node's owning-list reference before its next), + /// so a racing get_Next() dereferences list.head and throws + /// . Snapshotting under the lock is immune to that race. + /// + public T[] Snapshot() + { + lockSlim.EnterReadLock(); + try + { + var copy = new T[list.Count]; + list.CopyTo(copy, 0); + return copy; + } + finally + { + lockSlim.ExitReadLock(); + } + } + /// /// Gets the first node of the list under the read lock, or null if empty. /// From 5a4b6d78a01370a16476ea82cac0e7b260688ee1 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Roland=20B=C3=BCrgi?= Date: Mon, 6 Jul 2026 23:48:52 +0200 Subject: [PATCH 2/2] test(messaging): regression guard for the rule-chain dispatch race MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Hammers MessageHub dispatch while a background task disposes+re-registers rules (each Dispose is a rules.Remove), over 300 round-trips on a 100-rule chain. Proven: fails on the pre-fix raw-LinkedListNode.Next walk (the delivery NREs → the response never lands → 10s timeout) and passes deterministically with the snapshot fix. Addresses the Copilot review on #333. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../RuleChainConcurrentDisposeTest.cs | 81 +++++++++++++++++++ 1 file changed, 81 insertions(+) create mode 100644 test/MeshWeaver.Messaging.Hub.Test/RuleChainConcurrentDisposeTest.cs diff --git a/test/MeshWeaver.Messaging.Hub.Test/RuleChainConcurrentDisposeTest.cs b/test/MeshWeaver.Messaging.Hub.Test/RuleChainConcurrentDisposeTest.cs new file mode 100644 index 0000000000..0256269c01 --- /dev/null +++ b/test/MeshWeaver.Messaging.Hub.Test/RuleChainConcurrentDisposeTest.cs @@ -0,0 +1,81 @@ +using System; +using System.Collections.Generic; +using System.Reactive.Linq; +using System.Threading; +using System.Threading.Tasks; +using MeshWeaver.Fixture; +using Xunit; + +namespace MeshWeaver.Messaging.Hub.Test; + +/// +/// Regression guard for the rule-chain dispatch race. +/// dispatches a delivery by folding over its rule chain (a +/// ). It used to walk that chain via the raw +/// OUTSIDE the list's lock, so a +/// concurrent rules.Remove — a handler disposable firing during teardown / rapid handler +/// churn — invalidated the node mid-walk ( +/// nulls the removed node's owning-list reference before its next) and get_Next() +/// threw → the delivery failed → the request never got its +/// response. This test hammers dispatch while concurrently registering/disposing rules. +/// With the snapshot fix every dispatch completes cleanly (deterministic — no flakiness); +/// a revert to the raw-node walk NREs the delivery and the round-trips below time out. +/// +public class RuleChainConcurrentDisposeTest(ITestOutputHelper output) : HubTestBase(output) +{ + private record PingReq : IRequest; + private record PongEvent; + + protected override MessageHubConfiguration ConfigureHost(MessageHubConfiguration configuration) => + configuration.WithHandler((hub, request) => + { + hub.Post(new PongEvent(), o => o.ResponseFor(request)); + return request.Processed(); + }); + + [Fact(Timeout = 60_000)] + public async Task Dispatch_WhileRulesAreConcurrentlyRemoved_NeverFaults() + { + var host = GetHost(); + var target = CreateHostAddress(); + + // Fill the rule chain with many pass-through rules so a dispatch spends real time walking it, + // widening the window in which a concurrent Remove can invalidate the current node. + const int ruleCount = 100; + var disposables = new List(ruleCount); + for (var i = 0; i < ruleCount; i++) + disposables.Add(host.Register((d, _) => Observable.Return(d))); + + using var cts = new CancellationTokenSource(); + + // Background churn: a tight loop disposing + re-registering pass-through rules (each Dispose is + // a rules.Remove), racing the dispatch's chain walk — the exact remove-during-dispatch shape. + // Only this task touches `disposables`; the dispatch loop below never does. + var churn = Task.Run(() => + { + var n = 0; + while (!cts.IsCancellationRequested) + { + var idx = n++ % disposables.Count; + disposables[idx].Dispose(); + disposables[idx] = host.Register((d, _) => Observable.Return(d)); + } + }); + + try + { + // Many round-trips; each dispatch folds over the churning rule chain. With the fix all + // resolve; a raw-walk regression NREs the delivery so the response never lands → timeout. + for (var i = 0; i < 300; i++) + await host.Observe(new PingReq(), o => o.WithTarget(target)) + .Should().Within(10.Seconds()).Emit(); + } + finally + { + cts.Cancel(); + await churn; + foreach (var d in disposables) + d.Dispose(); + } + } +}