feat(mesh-phase5): node-local telemetry hub (snapshot-replay + drop-oldest fan-out)

Claude-Session: https://claude.ai/code/session_01GASWkNEi68FSCtvr6rLoEW
2026-07-23 15:08:56 -04:00
parent 53ae0100a2
commit 9882b1d250
4 changed files with 379 additions and 0 deletions
@@ -28,6 +28,7 @@ using ZB.MOM.WW.OtOpcUa.Runtime.Health;
 using ZB.MOM.WW.OtOpcUa.Runtime.Historian;
 using ZB.MOM.WW.OtOpcUa.Runtime.OpcUa;
 using ZB.MOM.WW.OtOpcUa.Runtime.Redundancy;
+using ZB.MOM.WW.OtOpcUa.Runtime.Telemetry;
 using ZB.MOM.WW.OtOpcUa.Runtime.VirtualTags;
 using ZB.MOM.WW.LocalDb;

@@ -70,6 +71,10 @@ public static class ServiceCollectionExtensions
        services.TryAddSingleton<IOpcUaAddressSpaceSink>(NullOpcUaAddressSpaceSink.Instance);
        services.TryAddSingleton<IServiceLevelPublisher>(NullServiceLevelPublisher.Instance);
        services.TryAddSingleton<IDriverHealthPublisher, AkkaDriverHealthPublisher>();
+        // Per-cluster mesh Phase 5: the node-local live-telemetry fan-out hub. Feeds this node's OWN
+        // telemetry (never DPS) to the Phase-5 gRPC streaming service central dials in on. AddOtOpcUaRuntime
+        // runs only inside Program.cs's hasDriver block, so this lands on driver-role nodes only.
+        services.TryAddSingleton<ITelemetryLocalHub, TelemetryLocalHub>();
        return services;
    }

@@ -0,0 +1,80 @@
+using System.Threading.Channels;
+using ZB.MOM.WW.OtOpcUa.Commons.Messages.Alerts;
+using ZB.MOM.WW.OtOpcUa.Commons.Messages.Drivers;
+using ZB.MOM.WW.OtOpcUa.Commons.Messages.Logging;
+
+namespace ZB.MOM.WW.OtOpcUa.Runtime.Telemetry;
+
+/// <summary>
+///     A single live-telemetry item flowing through the node-local hub. A closed union over the four
+///     wire-facing telemetry records — carried as the domain records themselves, NOT as proto (the
+///     gRPC projection happens at the streaming-service seam, one layer out).
+/// </summary>
+/// <remarks>
+///     Two of the four are <b>snapshot-style</b> (a last-value-per-key view: <see cref="Health"/> keyed
+///     by <c>DriverInstanceId</c>, <see cref="Resilience"/> keyed by <c>(DriverInstanceId, HostName)</c>)
+///     and are cached so a newly-attaching subscriber is primed immediately. The other two
+///     (<see cref="Alarm"/>, <see cref="Script"/>) are <b>append-style</b> logs and are forwarded live-only.
+/// </remarks>
+public abstract record TelemetryItem
+{
+    private TelemetryItem() { }
+
+    /// <summary>An append-style alarm transition. Live-forwarded only; never cached/replayed.</summary>
+    public sealed record Alarm(AlarmTransitionEvent E) : TelemetryItem;
+
+    /// <summary>An append-style script-log line. Live-forwarded only; never cached/replayed.</summary>
+    public sealed record Script(ScriptLogEntry E) : TelemetryItem;
+
+    /// <summary>A snapshot-style driver-health change, cached last-value per <c>DriverInstanceId</c>.</summary>
+    public sealed record Health(DriverHealthChanged E) : TelemetryItem;
+
+    /// <summary>A snapshot-style resilience-status change, cached last-value per <c>(DriverInstanceId, HostName)</c>.</summary>
+    public sealed record Resilience(DriverResilienceStatusChanged E) : TelemetryItem;
+}
+
+/// <summary>
+///     One attached reader on the <see cref="ITelemetryLocalHub"/>. Disposing detaches it from the hub
+///     and completes its channel; the reader first yields the cached snapshots, then live deltas.
+/// </summary>
+public interface ITelemetrySubscription : IDisposable
+{
+    /// <summary>The bounded reader this subscription drains. Completed on <see cref="IDisposable.Dispose"/>.</summary>
+    ChannelReader<TelemetryItem> Reader { get; }
+}
+
+/// <summary>
+///     Process-wide, node-local fan-out hub sitting between this node's own telemetry producers and the
+///     (Phase-5) gRPC streaming service central dials into.
+/// </summary>
+/// <remarks>
+///     <para>
+///         <b>CRITICAL INVARIANT — this hub carries ONLY this node's own telemetry.</b> It is fed
+///         directly, in-process, by the node's producers and MUST NEVER subscribe to cluster
+///         DistributedPubSub. On the single mesh, DPS delivers every node's events to every node; if the
+///         hub read DPS it would stream peers' events and central — which dials each node individually —
+///         would double-count.
+///     </para>
+///     <para>
+///         Fan-out is lossy under backpressure by design: each subscriber owns a bounded channel with
+///         <see cref="BoundedChannelFullMode.DropOldest"/>, so a slow consumer sheds its own oldest items
+///         and can never block the producer / node-actor threads.
+///     </para>
+/// </remarks>
+public interface ITelemetryLocalHub
+{
+    /// <summary>
+    ///     Fans <paramref name="item"/> to every currently-attached subscriber (non-blocking, drop-oldest
+    ///     on a full channel) and, for the two snapshot-style items, updates the last-value cache that
+    ///     primes future subscribers.
+    /// </summary>
+    void Emit(TelemetryItem item);
+
+    /// <summary>
+    ///     Attaches a new subscriber. Its reader first yields the cached snapshots (all cached
+    ///     <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>
+    ITelemetrySubscription Subscribe(int boundedCapacity);
+}
@@ -0,0 +1,121 @@
+using System.Collections.Concurrent;
+using System.Threading.Channels;
+
+namespace ZB.MOM.WW.OtOpcUa.Runtime.Telemetry;
+
+/// <summary>
+///     Default <see cref="ITelemetryLocalHub"/>: a process-wide singleton on driver-role nodes.
+/// </summary>
+/// <remarks>
+///     <para>
+///         <b>No DPS.</b> This hub is fed only by this node's own in-process producers (a later Phase-5
+///         task taps them). It never subscribes to cluster DistributedPubSub — see the invariant on
+///         <see cref="ITelemetryLocalHub"/>.
+///     </para>
+///     <para>
+///         <b>Snapshot-then-attach ordering (no lost delta).</b> <see cref="Emit"/> updates the snapshot
+///         cache and fans out under a single gate; <see cref="Subscribe"/> writes the cached snapshots
+///         into the new channel and only THEN adds it to the subscriber set — all under that same gate.
+///         Because both operations are serialized by the gate, an <see cref="Emit"/> racing a
+///         <see cref="Subscribe"/> is resolved one of two ways, both correct: it runs entirely before the
+///         subscribe (its value is in the snapshot, and it is not delivered live because the channel is
+///         not yet attached), or entirely after (it is delivered live, strictly after the snapshot). No
+///         item is dropped or duplicated across the boundary. Fan-out is a non-blocking
+///         <see cref="ChannelWriter{T}.TryWrite"/> into bounded/DropOldest channels, so holding the gate
+///         never blocks on a slow consumer.
+///     </para>
+/// </remarks>
+public sealed class TelemetryLocalHub : ITelemetryLocalHub
+{
+    private readonly object _gate = new();
+    private readonly ConcurrentDictionary<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).
+    private readonly ConcurrentDictionary<string, TelemetryItem.Health> _healthCache = new();
+    private readonly ConcurrentDictionary<(string InstanceId, string HostName), TelemetryItem.Resilience> _resilienceCache = new();
+
+    /// <inheritdoc />
+    public void Emit(TelemetryItem item)
+    {
+        ArgumentNullException.ThrowIfNull(item);
+
+        lock (_gate)
+        {
+            switch (item)
+            {
+                case TelemetryItem.Health h:
+                    _healthCache[h.E.DriverInstanceId] = h;
+                    break;
+                case TelemetryItem.Resilience r:
+                    _resilienceCache[(r.E.DriverInstanceId, r.E.HostName)] = r;
+                    break;
+                // Alarm / Script are append-style — never cached.
+            }
+
+            foreach (var channel in _subscribers.Values)
+                channel.Writer.TryWrite(item); // bounded + DropOldest ⇒ non-blocking, sheds oldest when full
+        }
+    }
+
+    /// <inheritdoc />
+    public ITelemetrySubscription Subscribe(int boundedCapacity)
+    {
+        if (boundedCapacity < 1)
+            throw new ArgumentOutOfRangeException(nameof(boundedCapacity), boundedCapacity, "Capacity must be positive.");
+
+        var channel = Channel.CreateBounded<TelemetryItem>(new BoundedChannelOptions(boundedCapacity)
+        {
+            FullMode = BoundedChannelFullMode.DropOldest,
+            SingleReader = true,
+            SingleWriter = false,
+        });
+
+        var id = Guid.NewGuid();
+        lock (_gate)
+        {
+            // Snapshot FIRST (Health then Resilience), attach SECOND — both under the gate so a
+            // concurrent Emit cannot slip a delta between the snapshot read and the attach.
+            foreach (var health in _healthCache.Values)
+                channel.Writer.TryWrite(health);
+            foreach (var resilience in _resilienceCache.Values)
+                channel.Writer.TryWrite(resilience);
+
+            _subscribers[id] = channel;
+        }
+
+        return new Subscription(this, id, channel);
+    }
+
+    private void Detach(Guid id)
+    {
+        lock (_gate)
+        {
+            if (_subscribers.TryRemove(id, out var channel))
+                channel.Writer.TryComplete();
+        }
+    }
+
+    private sealed class Subscription : ITelemetrySubscription
+    {
+        private readonly TelemetryLocalHub _hub;
+        private readonly Guid _id;
+        private int _disposed;
+
+        public Subscription(TelemetryLocalHub hub, Guid id, Channel<TelemetryItem> channel)
+        {
+            _hub = hub;
+            _id = id;
+            Reader = channel.Reader;
+        }
+
+        public ChannelReader<TelemetryItem> Reader { get; }
+
+        public void Dispose()
+        {
+            if (Interlocked.Exchange(ref _disposed, 1) != 0)
+                return; // idempotent
+            _hub.Detach(_id);
+        }
+    }
+}
@@ -0,0 +1,173 @@
+using Shouldly;
+using Xunit;
+using ZB.MOM.WW.OtOpcUa.Commons.Messages.Alerts;
+using ZB.MOM.WW.OtOpcUa.Commons.Messages.Drivers;
+using ZB.MOM.WW.OtOpcUa.Commons.Messages.Logging;
+using ZB.MOM.WW.OtOpcUa.Runtime.Telemetry;
+
+namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Telemetry;
+
+/// <summary>
+/// Verifies the node-local telemetry hub (per-cluster mesh Phase 5): snapshot-replay of the two
+/// snapshot-style channels (Health / Resilience), live-only forwarding of the append-style logs
+/// (Alarm / Script), independent bounded per-subscriber fan-out, DropOldest backpressure, and
+/// detach-on-dispose. The hub carries ONLY this node's own telemetry — it never touches DPS.
+/// </summary>
+public sealed class TelemetryLocalHubTests
+{
+    private static DriverHealthChanged Health(string instanceId, string state = "Healthy") =>
+        new("cluster-a", instanceId, state, null, null, 0, DateTime.UtcNow);
+
+    private static DriverResilienceStatusChanged Resilience(string instanceId, string host) =>
+        new(instanceId, host, false, 0, 0, null, DateTime.UtcNow, DateTime.UtcNow);
+
+    private static AlarmTransitionEvent Alarm(string id) =>
+        new(id, "Area/Line/Equip", "AlarmName", "Activated", 500, "msg", "system", DateTime.UtcNow);
+
+    private static ScriptLogEntry Script(string id) =>
+        new(id, "Information", "log", DateTime.UtcNow, null, null, null);
+
+    private static List<TelemetryItem> Drain(ITelemetrySubscription sub)
+    {
+        var items = new List<TelemetryItem>();
+        while (sub.Reader.TryRead(out var item))
+            items.Add(item);
+        return items;
+    }
+
+    [Fact]
+    public void New_subscriber_is_primed_with_cached_Health_snapshot()
+    {
+        var hub = new TelemetryLocalHub();
+        hub.Emit(new TelemetryItem.Health(Health("drv-1", "Faulted")));
+
+        using var sub = hub.Subscribe(boundedCapacity: 16);
+
+        var items = Drain(sub);
+        items.ShouldHaveSingleItem();
+        var health = items[0].ShouldBeOfType<TelemetryItem.Health>();
+        health.E.DriverInstanceId.ShouldBe("drv-1");
+        health.E.State.ShouldBe("Faulted");
+    }
+
+    [Fact]
+    public void New_subscriber_is_primed_with_cached_Resilience_snapshot()
+    {
+        var hub = new TelemetryLocalHub();
+        hub.Emit(new TelemetryItem.Resilience(Resilience("drv-1", "host-a")));
+
+        using var sub = hub.Subscribe(boundedCapacity: 16);
+
+        var items = Drain(sub);
+        items.ShouldHaveSingleItem();
+        var res = items[0].ShouldBeOfType<TelemetryItem.Resilience>();
+        res.E.DriverInstanceId.ShouldBe("drv-1");
+        res.E.HostName.ShouldBe("host-a");
+    }
+
+    [Fact]
+    public void Alarm_and_Script_are_not_replayed_to_a_new_subscriber()
+    {
+        var hub = new TelemetryLocalHub();
+        hub.Emit(new TelemetryItem.Alarm(Alarm("a-1")));
+        hub.Emit(new TelemetryItem.Script(Script("s-1")));
+
+        using var sub = hub.Subscribe(boundedCapacity: 16);
+
+        Drain(sub).ShouldBeEmpty();
+    }
+
+    [Fact]
+    public void Two_concurrent_subscribers_each_receive_a_post_subscribe_emit()
+    {
+        var hub = new TelemetryLocalHub();
+        using var s1 = hub.Subscribe(boundedCapacity: 16);
+        using var s2 = hub.Subscribe(boundedCapacity: 16);
+
+        hub.Emit(new TelemetryItem.Alarm(Alarm("a-1")));
+
+        Drain(s1).ShouldHaveSingleItem().ShouldBeOfType<TelemetryItem.Alarm>().E.AlarmId.ShouldBe("a-1");
+        Drain(s2).ShouldHaveSingleItem().ShouldBeOfType<TelemetryItem.Alarm>().E.AlarmId.ShouldBe("a-1");
+    }
+
+    [Fact]
+    public void Health_snapshot_keeps_only_the_latest_per_instance()
+    {
+        var hub = new TelemetryLocalHub();
+        hub.Emit(new TelemetryItem.Health(Health("drv-1", "Reconnecting")));
+        hub.Emit(new TelemetryItem.Health(Health("drv-1", "Healthy")));
+
+        using var sub = hub.Subscribe(boundedCapacity: 16);
+
+        var items = Drain(sub);
+        items.ShouldHaveSingleItem();
+        items[0].ShouldBeOfType<TelemetryItem.Health>().E.State.ShouldBe("Healthy");
+    }
+
+    [Fact]
+    public void Health_snapshot_keeps_distinct_instances()
+    {
+        var hub = new TelemetryLocalHub();
+        hub.Emit(new TelemetryItem.Health(Health("drv-1")));
+        hub.Emit(new TelemetryItem.Health(Health("drv-2")));
+
+        using var sub = hub.Subscribe(boundedCapacity: 16);
+
+        var ids = Drain(sub)
+            .OfType<TelemetryItem.Health>()
+            .Select(h => h.E.DriverInstanceId)
+            .OrderBy(x => x)
+            .ToArray();
+        ids.ShouldBe(new[] { "drv-1", "drv-2" });
+    }
+
+    [Fact]
+    public void Resilience_snapshot_keeps_only_latest_per_instance_and_host()
+    {
+        var hub = new TelemetryLocalHub();
+        // Same instance, two distinct hosts → two cache entries.
+        hub.Emit(new TelemetryItem.Resilience(Resilience("drv-1", "host-a")));
+        hub.Emit(new TelemetryItem.Resilience(Resilience("drv-1", "host-b")));
+        // Overwrite (drv-1, host-a).
+        hub.Emit(new TelemetryItem.Resilience(Resilience("drv-1", "host-a")));
+
+        using var sub = hub.Subscribe(boundedCapacity: 16);
+
+        var keys = Drain(sub)
+            .OfType<TelemetryItem.Resilience>()
+            .Select(r => (r.E.DriverInstanceId, r.E.HostName))
+            .OrderBy(x => x.HostName)
+            .ToArray();
+        keys.ShouldBe(new[] { ("drv-1", "host-a"), ("drv-1", "host-b") });
+    }
+
+    [Fact]
+    public void Bounded_channel_drops_oldest_not_newest_when_full()
+    {
+        var hub = new TelemetryLocalHub();
+        using var sub = hub.Subscribe(boundedCapacity: 2);
+
+        // Append-style (uncached) items so the snapshot doesn't consume capacity.
+        hub.Emit(new TelemetryItem.Alarm(Alarm("a-1")));
+        hub.Emit(new TelemetryItem.Alarm(Alarm("a-2")));
+        hub.Emit(new TelemetryItem.Alarm(Alarm("a-3")));
+
+        var ids = Drain(sub).OfType<TelemetryItem.Alarm>().Select(a => a.E.AlarmId).ToArray();
+        // a-1 (oldest) evicted; the two newest survive in order.
+        ids.ShouldBe(new[] { "a-2", "a-3" });
+    }
+
+    [Fact]
+    public void Dispose_detaches_the_subscriber_and_completes_its_channel()
+    {
+        var hub = new TelemetryLocalHub();
+        var sub = hub.Subscribe(boundedCapacity: 16);
+
+        sub.Dispose();
+
+        // A further Emit must neither throw nor deliver to the detached subscriber.
+        Should.NotThrow(() => hub.Emit(new TelemetryItem.Alarm(Alarm("a-1"))));
+        sub.Reader.Completion.IsCompleted.ShouldBeTrue();
+        sub.Reader.TryRead(out _).ShouldBeFalse();
+    }
+}