diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Telemetry/ITelemetryLocalHub.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Telemetry/ITelemetryLocalHub.cs index 127cb755..036b6f3c 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Telemetry/ITelemetryLocalHub.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Telemetry/ITelemetryLocalHub.cs @@ -75,6 +75,13 @@ public interface ITelemetryLocalHub /// , then all cached ), /// then live deltas — with no delta lost across the attach boundary. Dispose to detach. /// - /// Per-subscriber channel capacity (must be positive). + /// + /// Per-subscriber channel capacity (must be positive). The snapshot prelude is written into this + /// same bounded/DropOldest channel before returns, so + /// 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. + /// ITelemetrySubscription Subscribe(int boundedCapacity); } diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Telemetry/TelemetryLocalHub.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Telemetry/TelemetryLocalHub.cs index 6aaf9ba9..aa70c434 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Telemetry/TelemetryLocalHub.cs +++ b/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> _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> _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 _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(); } } diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Telemetry/TelemetryLocalHubTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Telemetry/TelemetryLocalHubTests.cs index 81613977..23eab83c 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Telemetry/TelemetryLocalHubTests.cs +++ b/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(() => hub.Subscribe(0)); + Should.Throw(() => 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(); + + // 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(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(); + // (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() {