fix(mesh-phase5): telemetry hub — plain Dictionary under gate, concurrency stress test, priming/eviction notes

Claude-Session: https://claude.ai/code/session_01GASWkNEi68FSCtvr6rLoEW

src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Telemetry/ITelemetryLocalHub.cs

@@ -75,6 +75,13 @@ public interface ITelemetryLocalHub
     ///     <see cref="TelemetryItem.Health"/>, then all cached <see cref="TelemetryItem.Resilience"/>),
     ///     then live deltas — with no delta lost across the attach boundary. Dispose to detach.
     /// </summary>
-    /// <param name="boundedCapacity">Per-subscriber channel capacity (must be positive).</param>
+    /// <param name="boundedCapacity">
+    ///     Per-subscriber channel capacity (must be positive). The snapshot prelude is written into this
+    ///     same bounded/DropOldest channel before <see cref="Subscribe"/> returns, so
+    ///     <paramref name="boundedCapacity"/> should comfortably exceed this node's live driver-instance +
+    ///     (instance, host) count — otherwise the earliest cached snapshots are silently evicted during
+    ///     priming before the reader ever drains them. The node-local cache holds only THIS node's
+    ///     instances (a handful), so a normal large capacity leaves ample headroom.
+    /// </param>
     ITelemetrySubscription Subscribe(int boundedCapacity);
 }

src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Telemetry/TelemetryLocalHub.cs

@@ -28,10 +28,18 @@ namespace ZB.MOM.WW.OtOpcUa.Runtime.Telemetry;
 public sealed class TelemetryLocalHub : ITelemetryLocalHub
 {
     private readonly object _gate = new();
-    private readonly ConcurrentDictionary<Guid, Channel<TelemetryItem>> _subscribers = new();
+
+    // Plain Dictionary, NOT ConcurrentDictionary: every access is already under _gate, and
+    // ConcurrentDictionary.Values would materialize a fresh List (and take its own internal lock) on
+    // every Emit — the hot node-actor-thread path. Under the gate a plain Dictionary is safe and
+    // allocation-free to iterate.
+    private readonly Dictionary<Guid, Channel<TelemetryItem>> _subscribers = new();
 
     // Snapshot last-value caches. Only ever mutated under _gate (kept ConcurrentDictionary so a
     // Subscribe reading them under the gate is trivially safe even against any future lock-free read).
+    // TODO(mesh-phase5+): evict snapshot cache entries on driver-instance removal (needs a
+    // driver-lifecycle hook); pre-production, accepted for now. Until then a decommissioned driver
+    // instance's last-known Health/Resilience state replays to new subscribers forever.
     private readonly ConcurrentDictionary<string, TelemetryItem.Health> _healthCache = new();
     private readonly ConcurrentDictionary<(string InstanceId, string HostName), TelemetryItem.Resilience> _resilienceCache = new();
 
@@ -91,7 +99,7 @@ public sealed class TelemetryLocalHub : ITelemetryLocalHub
     {
         lock (_gate)
         {
-            if (_subscribers.TryRemove(id, out var channel))
+            if (_subscribers.Remove(id, out var channel))
                 channel.Writer.TryComplete();
         }
     }

tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Telemetry/TelemetryLocalHubTests.cs

@@ -1,3 +1,4 @@
+using System.Collections.Concurrent;
 using Shouldly;
 using Xunit;
 using ZB.MOM.WW.OtOpcUa.Commons.Messages.Alerts;
@@ -78,7 +79,7 @@ public sealed class TelemetryLocalHubTests
     }
 
     [Fact]
-    public void Two_concurrent_subscribers_each_receive_a_post_subscribe_emit()
+    public void Multiple_subscribers_each_receive_their_own_copy_of_a_post_subscribe_emit()
     {
         var hub = new TelemetryLocalHub();
         using var s1 = hub.Subscribe(boundedCapacity: 16);
@@ -157,6 +158,81 @@ public sealed class TelemetryLocalHubTests
         ids.ShouldBe(new[] { "a-2", "a-3" });
     }
 
+    [Fact]
+    public void Subscribe_with_non_positive_capacity_throws()
+    {
+        var hub = new TelemetryLocalHub();
+        Should.Throw<ArgumentOutOfRangeException>(() => hub.Subscribe(0));
+        Should.Throw<ArgumentOutOfRangeException>(() => hub.Subscribe(-1));
+    }
+
+    [Fact]
+    public async Task Concurrent_emit_subscribe_dispose_is_safe_and_never_double_delivers_a_snapshot()
+    {
+        var hub = new TelemetryLocalHub();
+        var stop = new CancellationTokenSource(TimeSpan.FromSeconds(3));
+        var exceptions = new ConcurrentQueue<Exception>();
+
+        // Writers: hammer Emit with a mix of all four kinds across a small key space so the snapshot
+        // caches churn (same instance/host reused → last-value overwrites) while subscribers attach.
+        var writers = Enumerable.Range(0, 4).Select(w => Task.Run(() =>
+        {
+            var rnd = new Random(w * 7919 + 1);
+            try
+            {
+                var n = 0;
+                while (!stop.IsCancellationRequested)
+                {
+                    var instance = "drv-" + rnd.Next(0, 4);
+                    var host = "host-" + rnd.Next(0, 3);
+                    TelemetryItem item = (n++ % 4) switch
+                    {
+                        0 => new TelemetryItem.Health(Health(instance, "s" + n)),
+                        1 => new TelemetryItem.Resilience(Resilience(instance, host)),
+                        2 => new TelemetryItem.Alarm(Alarm("a-" + n)),
+                        _ => new TelemetryItem.Script(Script("s-" + n)),
+                    };
+                    hub.Emit(item);
+                }
+            }
+            catch (Exception ex) { exceptions.Enqueue(ex); }
+        })).ToArray();
+
+        // Subscriber/disposer churn: attach, drain, verify the exactly-once-across-attach invariant, dispose.
+        var churners = Enumerable.Range(0, 6).Select(_ => Task.Run(() =>
+        {
+            try
+            {
+                while (!stop.IsCancellationRequested)
+                {
+                    using var sub = hub.Subscribe(boundedCapacity: 4096);
+                    // The snapshot prelude is fully written before Subscribe returns, so any Health/
+                    // Resilience item read here that was NOT part of the prelude must be a live delta —
+                    // i.e. it arrived strictly after attach. The invariant we assert: for one snapshot
+                    // KEY, the hub never delivers the SAME snapshot instance both in the prelude and
+                    // again live. We approximate "same snapshot" by reference identity: a live re-emit
+                    // is always a freshly-allocated record, so a reference-equal duplicate would be a
+                    // genuine double-delivery of one cached object across the boundary.
+                    var seen = new HashSet<TelemetryItem>(ReferenceEqualityComparer.Instance);
+                    while (sub.Reader.TryRead(out var item))
+                    {
+                        // (c) every drained item is a valid, non-null TelemetryItem of a known kind.
+                        item.ShouldNotBeNull();
+                        item.ShouldBeAssignableTo<TelemetryItem>();
+                        // (b) no cached snapshot object delivered twice to THIS subscriber.
+                        seen.Add(item).ShouldBeTrue();
+                    }
+                }
+            }
+            catch (Exception ex) { exceptions.Enqueue(ex); }
+        })).ToArray();
+
+        await Task.WhenAll(writers.Concat(churners));
+
+        // (a) no thread threw.
+        exceptions.ShouldBeEmpty();
+    }
+
     [Fact]
     public void Dispose_detaches_the_subscriber_and_completes_its_channel()
     {