fix(messaging): snapshot rule chain — concurrent Remove NREs the dispatch walk (sync-hub flake)

src/MeshWeaver.Messaging.Hub/MessageHub.cs

@@ -675,7 +675,7 @@ private record TypeAndHandler(Type Type, AsyncDelivery? Action);
 
     private IObservable<IMessageDelivery> HandleMessageAsync(
         IMessageDelivery delivery,
-        LinkedListNode<AsyncDelivery> 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<IMessageDelivery> result = Observable.Return(delivery);
-        LinkedListNode<AsyncDelivery>? 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

src/MeshWeaver.Messaging.Hub/ThreadSafeLinkedList.cs

@@ -62,6 +62,30 @@ public void Remove(LinkedListNode<T> item)
         }
     }
 
+    /// <summary>
+    /// Returns a point-in-time copy of the values (head→tail) taken under the read lock.
+    /// <para>Callers that need to ITERATE the list MUST use this instead of walking
+    /// <see cref="LinkedListNode{T}.Next"/> off <see cref="First"/>: a raw <c>.Next</c> walk runs
+    /// outside the lock, and a concurrent <see cref="Remove"/> invalidates the node mid-walk
+    /// (<see cref="LinkedList{T}"/> nulls the node's owning-list reference before its <c>next</c>),
+    /// so a racing <c>get_Next()</c> dereferences <c>list.head</c> and throws
+    /// <see cref="NullReferenceException"/>. Snapshotting under the lock is immune to that race.</para>
+    /// </summary>
+    public T[] Snapshot()
+    {
+        lockSlim.EnterReadLock();
+        try
+        {
+            var copy = new T[list.Count];
+            list.CopyTo(copy, 0);
+            return copy;
+        }
+        finally
+        {
+            lockSlim.ExitReadLock();
+        }
+    }
+
     /// <summary>
     /// Gets the first node of the list under the read lock, or null if empty.
     /// </summary>

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;
+
+/// <summary>
+/// Regression guard for the rule-chain dispatch race.
+/// <para><see cref="MessageHub"/> dispatches a delivery by folding over its rule chain (a
+/// <see cref="ThreadSafeLinkedList{T}"/>). It used to walk that chain via the raw
+/// <see cref="System.Collections.Generic.LinkedListNode{T}.Next"/> OUTSIDE the list's lock, so a
+/// concurrent <c>rules.Remove</c> — a handler disposable firing during teardown / rapid handler
+/// churn — invalidated the node mid-walk (<see cref="System.Collections.Generic.LinkedList{T}"/>
+/// nulls the removed node's owning-list reference before its <c>next</c>) and <c>get_Next()</c>
+/// threw <see cref="NullReferenceException"/> → the delivery failed → the request never got its
+/// response. This test hammers dispatch while concurrently registering/disposing rules.</para>
+/// <para>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.</para>
+/// </summary>
+public class RuleChainConcurrentDisposeTest(ITestOutputHelper output) : HubTestBase(output)
+{
+    private record PingReq : IRequest<PongEvent>;
+    private record PongEvent;
+
+    protected override MessageHubConfiguration ConfigureHost(MessageHubConfiguration configuration) =>
+        configuration.WithHandler<PingReq>((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<IDisposable>(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();
+        }
+    }
+}