diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Browse/GalaxyDiscoverer.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Browse/GalaxyDiscoverer.cs index 6242351c..6e3bc5f8 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Browse/GalaxyDiscoverer.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Browse/GalaxyDiscoverer.cs @@ -61,8 +61,12 @@ public sealed class GalaxyDiscoverer /// /// The address space builder to populate with discovery results. /// The cancellation token for the operation. - /// A task that represents the asynchronous operation. - public async Task DiscoverAsync(IAddressSpaceBuilder builder, CancellationToken cancellationToken) + /// + /// The hierarchy this pass fetched, so a caller that needs the object set for something + /// else (the driver selects its $WinPlatform / $AppEngine probe targets from + /// it) reuses the one sweep instead of asking the gateway for the full galaxy twice. + /// + public async Task> DiscoverAsync(IAddressSpaceBuilder builder, CancellationToken cancellationToken) { ArgumentNullException.ThrowIfNull(builder); var objects = await _source.GetHierarchyAsync(cancellationToken).ConfigureAwait(false); @@ -132,6 +136,8 @@ public sealed class GalaxyDiscoverer } } } + + return objects; } // Resolve (and memoise) the folder for a gobject, recursively creating its parent's folder diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/GalaxyDriver.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/GalaxyDriver.cs index a75ba588..f2768291 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/GalaxyDriver.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/GalaxyDriver.cs @@ -454,8 +454,11 @@ public sealed class GalaxyDriver } } + // The watcher publishes its probe handles into the driver's registry (reserved id) so the + // EventPump's fan-out reaches OnPumpDataChange for ScanState events — without that every + // probe event was dropped before it got anywhere near the watcher. private PerPlatformProbeWatcher BuildProbeWatcher(IGalaxySubscriber subscriber) => - new(subscriber, _hostStatuses, _logger, + new(subscriber, _hostStatuses, _subscriptions, _logger, bufferedUpdateIntervalMs: _options.MxAccess.PublishingIntervalMs); /// @@ -738,6 +741,15 @@ public sealed class GalaxyDriver var tagCount = 0; foreach (var (subscriptionId, oldBindings) in entries) { + if (subscriptionId == SubscriptionRegistry.ProbeSubscriptionId && _probeWatcher is { } probeWatcher) + { + // The probe watcher owns its handles: it re-subscribes its platforms, records the + // fresh handles and rebinds the reserved registry entry itself, so ScanState host + // status keeps updating after a reconnect (it used to go permanently deaf). + tagCount += await probeWatcher.ResubscribeAsync(cancellationToken).ConfigureAwait(false); + continue; + } + var refs = oldBindings .Select(b => b.FullReference) .Distinct(StringComparer.OrdinalIgnoreCase) @@ -1058,11 +1070,12 @@ public sealed class GalaxyDriver // keyed by RawPath, matching what WriteAsync resolves against. var discoverer = new GalaxyDiscoverer( source, attributeRef => _rawTagByAttributeRef.GetValueOrDefault(attributeRef)); - await discoverer.DiscoverAsync(capturingBuilder, cancellationToken).ConfigureAwait(false); + // One sweep: the discoverer hands back the hierarchy it materialised, and the platform + // selection below reads that instead of asking the gateway for the whole galaxy again. + var hierarchy = await discoverer.DiscoverAsync(capturingBuilder, cancellationToken).ConfigureAwait(false); if (_probeWatcher is not null) { - var hierarchy = await source.GetHierarchyAsync(cancellationToken).ConfigureAwait(false); var platforms = hierarchy .Where(o => o.TemplateChain.Any(t => string.Equals(t, "$WinPlatform", StringComparison.OrdinalIgnoreCase) @@ -1744,6 +1757,24 @@ public sealed class GalaxyDriver /// private void OnPumpDataChange(object? sender, DataChangeEventArgs args) { + // Anything the gateway pushes on the stream proves it is alive and serving this session — + // the comms-loss watchdog's evidence (G-11: the stamp used to move only at init and on a + // supervisor Healthy transition, so a 90 s Degraded tripped a re-init under flowing data). + StampLiveness(); + + // Probe traffic rides the reserved registry entry and stays in the attributeRef domain — + // ScanState attributes are discovered platform attributes, never authored RawPaths. It is + // the watcher's alone: the host never subscribed it and must not see it on OnDataChange. + if (args.SubscriptionHandle is GalaxySubscriptionHandle { SubscriptionId: SubscriptionRegistry.ProbeSubscriptionId }) + { + // The probe decoder takes a raw quality byte. Recover it via the canonical + // StatusCodeMap.ToQualityCategoryByte helper so the mapping lives in one + // place next to its inverse (FromQualityByte) and cannot desync silently. + var qualityByte = StatusCodeMap.ToQualityCategoryByte(args.Snapshot.StatusCode); + _probeWatcher?.OnProbeValueChanged(args.FullReference, args.Snapshot.Value, qualityByte); + return; + } + // v3: the pump fans out keyed by the dialled attributeRef; the server correlates values to nodes // by RawPath (the node's FullName). Reverse-map before surfacing to public OnDataChange consumers. // Identity on a miss, so a live-browse tag surfaces under its attributeRef unchanged. @@ -1754,18 +1785,25 @@ public sealed class GalaxyDriver publicArgs = args with { FullReference = rawPath }; } OnDataChange?.Invoke(this, publicArgs); + } - // Probe watching stays in the attributeRef domain — ScanState attributes are discovered platform - // attributes, never authored RawPaths, so use the untranslated reference here. - if (_probeWatcher is not null - && args.FullReference.EndsWith(PerPlatformProbeWatcher.ProbeSuffix, StringComparison.OrdinalIgnoreCase)) - { - // The probe decoder takes a raw quality byte. Recover it via the canonical - // StatusCodeMap.ToQualityCategoryByte helper so the mapping lives in one - // place next to its inverse (FromQualityByte) and cannot desync silently. - var qualityByte = StatusCodeMap.ToQualityCategoryByte(args.Snapshot.StatusCode); - _probeWatcher.OnProbeValueChanged(args.FullReference, args.Snapshot.Value, qualityByte); - } + /// Minimum gap between two LastSuccessfulRead stamps from stream data. Internal so tests can widen it. + internal TimeSpan LivenessStampThrottle { get; set; } = TimeSpan.FromSeconds(1); + + private long _lastLivenessStampTicks; + + /// + /// Advance LastSuccessfulRead from inbound stream data, at most once per + /// — this runs on the pump's dispatch loop for every + /// event, so a 50k-tag galaxy must not pay a health-record allocation per value. + /// + private void StampLiveness() + { + var now = DateTime.UtcNow; + var last = Volatile.Read(ref _lastLivenessStampTicks); + if (now.Ticks - last < LivenessStampThrottle.Ticks) return; + Volatile.Write(ref _lastLivenessStampTicks, now.Ticks); + _health = _health with { LastSuccessfulRead = now }; } /// diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Health/PerPlatformProbeWatcher.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Health/PerPlatformProbeWatcher.cs index 12afb837..ae4c41ac 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Health/PerPlatformProbeWatcher.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Health/PerPlatformProbeWatcher.cs @@ -38,6 +38,12 @@ public sealed class PerPlatformProbeWatcher : IDisposable private readonly ILogger _logger; private readonly int _bufferedUpdateIntervalMs; + // The driver's fan-out registry. The EventPump resolves every inbound event through it, so + // the probe handles must be registered there (under SubscriptionRegistry.ProbeSubscriptionId) + // or every ScanState event is dropped as "stale after unsubscribe". Null when the watcher is + // driven directly (unit tests of the decoding table). + private readonly SubscriptionRegistry? _registry; + // Tracked platform → gw item handle. Item handle 0 means the gw rejected the subscribe; // we keep the entry so SyncPlatformsAsync doesn't try to subscribe it again on every call. private readonly ConcurrentDictionary _itemHandlesByPlatform = @@ -55,9 +61,30 @@ public sealed class PerPlatformProbeWatcher : IDisposable HostStatusAggregator aggregator, ILogger? logger = null, int bufferedUpdateIntervalMs = 0) + : this(subscriber, aggregator, registry: null, logger, bufferedUpdateIntervalMs) + { + } + + /// + /// Driver-side ctor: the watcher publishes its probe handles into + /// under so the + /// fans ScanState events out to . + /// + /// The Galaxy subscriber for managing probe subscriptions. + /// The host status aggregator for tracking platform connectivity. + /// The driver's subscription registry; null to keep the handles private. + /// Optional logger for diagnostic messages. + /// Buffered update interval in milliseconds; must be >= 0. + internal PerPlatformProbeWatcher( + IGalaxySubscriber subscriber, + HostStatusAggregator aggregator, + SubscriptionRegistry? registry, + ILogger? logger = null, + int bufferedUpdateIntervalMs = 0) { _subscriber = subscriber ?? throw new ArgumentNullException(nameof(subscriber)); _aggregator = aggregator ?? throw new ArgumentNullException(nameof(aggregator)); + _registry = registry; _logger = logger ?? NullLogger.Instance; if (bufferedUpdateIntervalMs < 0) { @@ -67,6 +94,68 @@ public sealed class PerPlatformProbeWatcher : IDisposable _bufferedUpdateIntervalMs = bufferedUpdateIntervalMs; } + /// Current platform → probe-handle bindings, as the registry sees them. Test seam. + internal IReadOnlyList CurrentBindings => BuildBindings(); + + private IReadOnlyList BuildBindings() => + [.. _itemHandlesByPlatform.Select(kvp => new TagBinding(kvp.Key + ProbeSuffix, kvp.Value))]; + + /// Mirror the tracked handle set into the driver registry (no-op without one). + private void PublishBindings() + { + if (_registry is null) return; + var bindings = BuildBindings(); + if (bindings.Count == 0) _registry.Remove(SubscriptionRegistry.ProbeSubscriptionId); + else _registry.Upsert(SubscriptionRegistry.ProbeSubscriptionId, bindings); + } + + /// + /// Re-issue SubscribeBulk for every tracked platform and replace the handles — the + /// reconnect replay path, where the gateway session was recreated and every pre-reconnect + /// handle is dead. Platforms the gateway now rejects keep a 0 handle (no events expected). + /// + /// A cancellation token to cancel the operation. + /// The number of probe references re-subscribed. + public async Task ResubscribeAsync(CancellationToken cancellationToken) + { + ObjectDisposedException.ThrowIf(_disposed, this); + var platforms = _itemHandlesByPlatform.Keys.ToArray(); + if (platforms.Length == 0) return 0; + + await SubscribeAndRecordAsync(platforms, cancellationToken).ConfigureAwait(false); + return platforms.Length; + } + + private async Task SubscribeAndRecordAsync(IReadOnlyList platforms, CancellationToken cancellationToken) + { + var probeAddresses = platforms.Select(p => p + ProbeSuffix).ToArray(); + // PR 6.3 — use the configured bufferedUpdateIntervalMs (defaults to 0 = gw cadence + // when the driver hasn't overridden MxAccess.PublishingIntervalMs). Probe ScanState + // changes are rare so a coarser interval is usually fine; deployments that need + // tighter health visibility can dial it down through GalaxyDriverOptions. + var results = await _subscriber.SubscribeBulkAsync( + probeAddresses, _bufferedUpdateIntervalMs, cancellationToken).ConfigureAwait(false); + + for (var i = 0; i < platforms.Count; i++) + { + var platform = platforms[i]; + var match = results.FirstOrDefault(r => string.Equals( + r.TagAddress, probeAddresses[i], StringComparison.OrdinalIgnoreCase)); + + var itemHandle = match is { WasSuccessful: true } ? match.ItemHandle : 0; + _itemHandlesByPlatform[platform] = itemHandle; + + if (itemHandle <= 0) + { + _logger.LogWarning( + "PerPlatformProbeWatcher subscribe failed for {Platform}: {Error}", + platform, match?.ErrorMessage ?? ""); + } + } + + PublishBindings(); + } + /// Snapshot of platform tag names currently watched. public IReadOnlyCollection WatchedPlatforms => [.. _itemHandlesByPlatform.Keys]; @@ -109,6 +198,8 @@ public sealed class PerPlatformProbeWatcher : IDisposable if (toRemove.Count > 0) { + // Drop the dead handles from the fan-out map before the (slow, fallible) unsubscribe. + PublishBindings(); try { await _subscriber.UnsubscribeBulkAsync( @@ -124,30 +215,7 @@ public sealed class PerPlatformProbeWatcher : IDisposable if (toAdd.Count == 0) return; - var probeAddresses = toAdd.Select(p => p + ProbeSuffix).ToArray(); - // PR 6.3 — use the configured bufferedUpdateIntervalMs (defaults to 0 = gw cadence - // when the driver hasn't overridden MxAccess.PublishingIntervalMs). Probe ScanState - // changes are rare so a coarser interval is usually fine; deployments that need - // tighter health visibility can dial it down through GalaxyDriverOptions. - var results = await _subscriber.SubscribeBulkAsync( - probeAddresses, _bufferedUpdateIntervalMs, cancellationToken).ConfigureAwait(false); - - for (var i = 0; i < toAdd.Count; i++) - { - var platform = toAdd[i]; - var match = results.FirstOrDefault(r => string.Equals( - r.TagAddress, probeAddresses[i], StringComparison.OrdinalIgnoreCase)); - - var itemHandle = match is { WasSuccessful: true } ? match.ItemHandle : 0; - _itemHandlesByPlatform[platform] = itemHandle; - - if (itemHandle <= 0) - { - _logger.LogWarning( - "PerPlatformProbeWatcher subscribe failed for {Platform}: {Error}", - platform, match?.ErrorMessage ?? ""); - } - } + await SubscribeAndRecordAsync(toAdd, cancellationToken).ConfigureAwait(false); } /// @@ -211,6 +279,7 @@ public sealed class PerPlatformProbeWatcher : IDisposable var liveHandles = _itemHandlesByPlatform.Values.Where(h => h > 0).ToArray(); _itemHandlesByPlatform.Clear(); + _registry?.Remove(SubscriptionRegistry.ProbeSubscriptionId); if (liveHandles.Length > 0) { try { await _subscriber.UnsubscribeBulkAsync(liveHandles, cancellationToken).ConfigureAwait(false); } diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/EventPump.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/EventPump.cs index e39385b4..7eb95234 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/EventPump.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/EventPump.cs @@ -285,14 +285,19 @@ internal sealed class EventPump : IAsyncDisposable EventsDispatched.Add(1, _clientTag); } } - catch (OperationCanceledException) when (ct.IsCancellationRequested) + catch (Exception) when (ct.IsCancellationRequested) { // Clean shutdown. } catch (Exception ex) { + // The read loop is still draining the stream into a channel nobody reads: the pump is + // as dead as after a transport fault, but used to raise nothing — health stayed + // Healthy and the comms-loss watchdog never tripped (a clean termination evading the + // watchdog). Report it as a stream fault so the supervisor recreates the pump. _logger.LogWarning(ex, - "Galaxy EventPump dispatch loop ended with an exception — events past this point will be lost until restart."); + "Galaxy EventPump dispatch loop ended with an exception — notifying reconnect supervisor so the pump is recreated."); + NotifyStreamFault(ex); } } diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/SubscriptionRegistry.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/SubscriptionRegistry.cs index 9dbd7755..1a6ac2ff 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/SubscriptionRegistry.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/SubscriptionRegistry.cs @@ -18,10 +18,39 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Runtime; /// internal sealed class SubscriptionRegistry { + /// + /// Reserved subscription id for the per-platform ScanState probe handles + /// (PerPlatformProbeWatcher). Driver subscription ids are positive and + /// monotonic (), so a negative id can never collide. + /// The probe handles used to live only in the watcher's own map, so the + /// — which resolves subscribers through THIS registry — + /// dropped every probe event as "stale after unsubscribe" and per-platform host + /// status never updated. + /// + public const long ProbeSubscriptionId = -1; + private readonly ConcurrentDictionary _bySubscriptionId = new(); private readonly ConcurrentDictionary> _subscribersByItemHandle = new(); private long _nextSubscriptionId; + /// True when is currently registered. + /// The subscription identifier. + /// Whether the id is tracked. + public bool Contains(long subscriptionId) => _bySubscriptionId.ContainsKey(subscriptionId); + + /// + /// when the id is already tracked (so its stale reverse-map entries + /// are dropped), otherwise. Used by the probe watcher, whose single + /// reserved subscription is re-issued on every platform-set change. + /// + /// The subscription identifier. + /// The tag bindings for the subscription. + public void Upsert(long subscriptionId, IReadOnlyList bindings) + { + if (Contains(subscriptionId)) Rebind(subscriptionId, bindings); + else Register(subscriptionId, bindings); + } + /// Gets the number of tracked subscriptions. public int TrackedSubscriptionCount => _bySubscriptionId.Count; /// Gets the number of tracked item handles. diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Health/PerPlatformProbeWatcherRegistryTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Health/PerPlatformProbeWatcherRegistryTests.cs new file mode 100644 index 00000000..511bb696 --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Health/PerPlatformProbeWatcherRegistryTests.cs @@ -0,0 +1,112 @@ +using System.Runtime.CompilerServices; +using Shouldly; +using Xunit; +using ZB.MOM.WW.MxGateway.Contracts.Proto; +using ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Health; +using ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Runtime; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests.Health; + +/// +/// Review 2026-10-21 G-4: the watcher's probe handles must live in the driver's +/// under the reserved id, follow the platform set, be +/// rebound on , and leave on stop. +/// +public sealed class PerPlatformProbeWatcherRegistryTests +{ + private sealed class FakeSubscriber : IGalaxySubscriber + { + private int _next; + public List Unsubscribes { get; } = []; + + public Task> SubscribeBulkAsync( + IReadOnlyList fullReferences, int bufferedUpdateIntervalMs, CancellationToken cancellationToken) + => Task.FromResult>( + [.. fullReferences.Select(r => new SubscribeResult { TagAddress = r, ItemHandle = ++_next, WasSuccessful = true })]); + + public Task UnsubscribeBulkAsync(IReadOnlyList itemHandles, CancellationToken cancellationToken) + { + Unsubscribes.Add([.. itemHandles]); + return Task.CompletedTask; + } + + public async IAsyncEnumerable StreamEventsAsync([EnumeratorCancellation] CancellationToken cancellationToken) + { + await Task.Delay(Timeout.Infinite, cancellationToken); + yield break; + } + } + + private static (long Id, string Ref)[] Resolve(SubscriptionRegistry registry, int handle) => + [.. registry.ResolveSubscribers(handle)]; + + /// Synced platforms resolve through the registry under the reserved id; removed ones stop resolving. + [Fact] + public async Task Sync_registers_probe_handles_under_the_reserved_id_and_tracks_removals() + { + var registry = new SubscriptionRegistry(); + using var watcher = new PerPlatformProbeWatcher(new FakeSubscriber(), new HostStatusAggregator(), registry); + + await watcher.SyncPlatformsAsync(["A", "B"], CancellationToken.None); + + registry.Contains(SubscriptionRegistry.ProbeSubscriptionId).ShouldBeTrue(); + var bindings = watcher.CurrentBindings; + bindings.Count.ShouldBe(2); + foreach (var b in bindings) + { + Resolve(registry, b.ItemHandle).ShouldBe([(SubscriptionRegistry.ProbeSubscriptionId, b.FullReference)]); + } + + var bHandle = bindings.Single(b => b.FullReference == "B.ScanState").ItemHandle; + await watcher.SyncPlatformsAsync(["A"], CancellationToken.None); + Resolve(registry, bHandle).ShouldBeEmpty("a removed platform's handle must leave the fan-out map"); + var aHandle = bindings.Single(b => b.FullReference == "A.ScanState").ItemHandle; + Resolve(registry, aHandle).ShouldHaveSingleItem("positive control: the kept platform still resolves"); + } + + /// Resubscribe replaces the handles and rebinds: old handles stop resolving, new ones resolve. + [Fact] + public async Task Resubscribe_rebinds_the_registry_to_the_fresh_handles() + { + var registry = new SubscriptionRegistry(); + using var watcher = new PerPlatformProbeWatcher(new FakeSubscriber(), new HostStatusAggregator(), registry); + await watcher.SyncPlatformsAsync(["A"], CancellationToken.None); + var oldHandle = watcher.CurrentBindings.Single().ItemHandle; + + (await watcher.ResubscribeAsync(CancellationToken.None)).ShouldBe(1); + + var newHandle = watcher.CurrentBindings.Single().ItemHandle; + newHandle.ShouldNotBe(oldHandle); + Resolve(registry, oldHandle).ShouldBeEmpty(); + Resolve(registry, newHandle).ShouldBe([(SubscriptionRegistry.ProbeSubscriptionId, "A.ScanState")]); + registry.TrackedSubscriptionCount.ShouldBe(1); + } + + /// StopAsync unsubscribes the live handles and removes the reserved entry. + [Fact] + public async Task StopAsync_unsubscribes_and_removes_the_reserved_entry() + { + var registry = new SubscriptionRegistry(); + var subscriber = new FakeSubscriber(); + var watcher = new PerPlatformProbeWatcher(subscriber, new HostStatusAggregator(), registry); + await watcher.SyncPlatformsAsync(["A", "B"], CancellationToken.None); + var handles = watcher.CurrentBindings.Select(b => b.ItemHandle).OrderBy(h => h).ToArray(); + + await watcher.StopAsync(CancellationToken.None); + + subscriber.Unsubscribes.ShouldHaveSingleItem().OrderBy(h => h).ShouldBe(handles); + registry.Contains(SubscriptionRegistry.ProbeSubscriptionId).ShouldBeFalse(); + registry.TrackedItemHandleCount.ShouldBe(0); + await Should.NotThrowAsync(() => watcher.StopAsync(CancellationToken.None)); // idempotent + } + + /// A watcher without a registry (the decoding-table unit tests) still works and registers nothing. + [Fact] + public async Task Without_a_registry_the_watcher_keeps_its_handles_private() + { + using var watcher = new PerPlatformProbeWatcher(new FakeSubscriber(), new HostStatusAggregator()); + await watcher.SyncPlatformsAsync(["A"], CancellationToken.None); + watcher.CurrentBindings.ShouldHaveSingleItem(); + (await watcher.ResubscribeAsync(CancellationToken.None)).ShouldBe(1); + } +} diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/EventPumpDispatchFaultTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/EventPumpDispatchFaultTests.cs new file mode 100644 index 00000000..8d48d6d5 --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/EventPumpDispatchFaultTests.cs @@ -0,0 +1,96 @@ +using System.Runtime.CompilerServices; +using System.Threading.Channels; +using Google.Protobuf.WellKnownTypes; +using Shouldly; +using Xunit; +using ZB.MOM.WW.MxGateway.Contracts.Proto; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; +using ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Runtime; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests.Runtime; + +/// +/// Review 2026-10-21 G-10: a dispatch-loop exit left the read loop draining into a channel +/// nobody read, with health Healthy and nothing telling the supervisor — a clean termination +/// evading the watchdog. The exit now reports a stream fault. The fault is injected through +/// the pump's handle factory, the one dispatch-side seam that runs outside the per-handler +/// isolation. +/// +public sealed class EventPumpDispatchFaultTests +{ + private const int WaitMs = 10_000; + + private sealed class ChannelSubscriber : IGalaxySubscriber + { + private readonly Channel _events = Channel.CreateUnbounded(); + + public Task> SubscribeBulkAsync( + IReadOnlyList fullReferences, int bufferedUpdateIntervalMs, CancellationToken cancellationToken) + => Task.FromResult>([]); + + public Task UnsubscribeBulkAsync(IReadOnlyList itemHandles, CancellationToken cancellationToken) => Task.CompletedTask; + + public async IAsyncEnumerable StreamEventsAsync([EnumeratorCancellation] CancellationToken cancellationToken) + { + await foreach (var ev in _events.Reader.ReadAllAsync(cancellationToken).ConfigureAwait(false)) + { + yield return ev; + } + } + + public ValueTask EmitAsync(int handle) => _events.Writer.WriteAsync(new MxEvent + { + Family = MxEventFamily.OnDataChange, + ItemHandle = handle, + Value = new MxValue { DoubleValue = 1 }, + Quality = 192, + SourceTimestamp = Timestamp.FromDateTime(DateTime.UtcNow), + }); + } + + /// A dispatch-loop exception reports a stream fault carrying that exception. + [Fact] + public async Task Dispatch_loop_exit_reports_a_stream_fault() + { + var subscriber = new ChannelSubscriber(); + var registry = new SubscriptionRegistry(); + registry.Register(1, [new TagBinding("Tank01.Level", 7)]); + var fault = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + + await using var pump = new EventPump( + subscriber, registry, + handleFactory: _ => throw new InvalidDataException("synthetic dispatch failure"), + channelCapacity: 4, clientName: "DispatchFault", + onStreamFault: ex => fault.TrySetResult(ex)); + pump.Start(); + + await subscriber.EmitAsync(7); + + (await Task.WhenAny(fault.Task, Task.Delay(WaitMs))).ShouldBe(fault.Task, + "a dispatch-loop exit must be reported as a stream fault"); + (await fault.Task).ShouldBeOfType(); + } + + /// Positive control: the same event with a working handle factory dispatches and reports no fault. + [Fact] + public async Task Same_event_with_a_working_factory_dispatches_without_a_fault() + { + var subscriber = new ChannelSubscriber(); + var registry = new SubscriptionRegistry(); + registry.Register(1, [new TagBinding("Tank01.Level", 7)]); + var faulted = false; + var delivered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + + await using var pump = new EventPump( + subscriber, registry, channelCapacity: 4, clientName: "DispatchFault", + onStreamFault: _ => faulted = true); + pump.OnDataChange += (_, e) => delivered.TrySetResult(e); + pump.Start(); + + await subscriber.EmitAsync(7); + + (await Task.WhenAny(delivered.Task, Task.Delay(WaitMs))).ShouldBe(delivered.Task); + (await delivered.Task).FullReference.ShouldBe("Tank01.Level"); + faulted.ShouldBeFalse(); + } +} diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/GalaxyDriverProbeWiringTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/GalaxyDriverProbeWiringTests.cs new file mode 100644 index 00000000..af842c93 --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/GalaxyDriverProbeWiringTests.cs @@ -0,0 +1,217 @@ +using System.Runtime.CompilerServices; +using System.Threading.Channels; +using Google.Protobuf.WellKnownTypes; +using Shouldly; +using Xunit; +using ZB.MOM.WW.MxGateway.Contracts.Proto; +using ZB.MOM.WW.MxGateway.Contracts.Proto.Galaxy; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; +using ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Browse; +using ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Config; +using ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Runtime; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests.Runtime; + +/// +/// Review 2026-10-21 G-4 / G-7 / G-11: the per-platform ScanState probing was structurally +/// dead (its handles were never in the registry the pump resolves through, and a reconnect +/// never re-subscribed them), discovery fetched the full galaxy twice, and +/// LastSuccessfulRead never moved on data. These drive the real driver → pump → watcher +/// → aggregator path with a fake subscriber and a fake hierarchy. +/// +public sealed class GalaxyDriverProbeWiringTests +{ + private const int WaitMs = 10_000; + + private static GalaxyDriverOptions Opts() => new( + new GalaxyGatewayOptions("https://mxgw.test:5001", "key"), + new GalaxyMxAccessOptions("ProbeWiring"), + new GalaxyRepositoryOptions(WatchDeployEvents: false), + new GalaxyReconnectOptions()); + + private sealed class CountingHierarchySource : IGalaxyHierarchySource + { + public int Calls; + + public Task> GetHierarchyAsync(CancellationToken cancellationToken) + { + Interlocked.Increment(ref Calls); + var platform = new GalaxyObject { GobjectId = 1, TagName = "Plat01", ContainedName = "Plat01" }; + platform.TemplateChain.Add("$WinPlatform"); + var tank = new GalaxyObject { GobjectId = 2, TagName = "Tank01", ContainedName = "Tank01", ParentGobjectId = 1 }; + tank.TemplateChain.Add("$UserDefined"); + tank.Attributes.Add(new GalaxyAttribute { AttributeName = "Level", FullTagReference = "Tank01.Level", MxDataType = 5 }); + return Task.FromResult>([platform, tank]); + } + } + + private sealed class StreamingSubscriber : IGalaxySubscriber + { + private readonly Channel _events = Channel.CreateUnbounded(); + private int _nextHandle; + + public List Subscribed { get; } = []; + public Dictionary HandleByRef { get; } = new(StringComparer.OrdinalIgnoreCase); + + public Task> SubscribeBulkAsync( + IReadOnlyList fullReferences, int bufferedUpdateIntervalMs, CancellationToken cancellationToken) + { + var results = new List(); + lock (Subscribed) + { + foreach (var r in fullReferences) + { + var h = Interlocked.Increment(ref _nextHandle); + Subscribed.Add(r); + HandleByRef[r] = h; + results.Add(new SubscribeResult { TagAddress = r, ItemHandle = h, WasSuccessful = true }); + } + } + return Task.FromResult>(results); + } + + public Task UnsubscribeBulkAsync(IReadOnlyList itemHandles, CancellationToken cancellationToken) => Task.CompletedTask; + + public async IAsyncEnumerable StreamEventsAsync([EnumeratorCancellation] CancellationToken cancellationToken) + { + await foreach (var ev in _events.Reader.ReadAllAsync(cancellationToken).ConfigureAwait(false)) + { + yield return ev; + } + } + + public ValueTask EmitBoolAsync(int handle, bool value) => _events.Writer.WriteAsync(new MxEvent + { + Family = MxEventFamily.OnDataChange, + ItemHandle = handle, + Value = new MxValue { BoolValue = value }, + Quality = 192, + SourceTimestamp = Timestamp.FromDateTime(DateTime.UtcNow), + }); + + public ValueTask EmitDoubleAsync(int handle, double value) => _events.Writer.WriteAsync(new MxEvent + { + Family = MxEventFamily.OnDataChange, + ItemHandle = handle, + Value = new MxValue { DoubleValue = value }, + Quality = 192, + SourceTimestamp = Timestamp.FromDateTime(DateTime.UtcNow), + }); + } + + private sealed class NullBuilder : IAddressSpaceBuilder + { + public IAddressSpaceBuilder Folder(string browseName, string displayName) => this; + public IVariableHandle Variable(string browseName, string displayName, DriverAttributeInfo attributeInfo) => new Handle(attributeInfo.FullName); + public void AddProperty(string browseName, DriverDataType dataType, object? value) { } + private sealed class Handle(string fullReference) : IVariableHandle + { + public string FullReference { get; } = fullReference; + public IAlarmConditionSink MarkAsAlarmCondition(AlarmConditionInfo alarmInfo) => new NoopSink(); + } + private sealed class NoopSink : IAlarmConditionSink + { + public void OnTransition(AlarmEventArgs args) { } + } + } + + private static async Task WaitUntilAsync(Func condition, string because) + { + var deadline = DateTime.UtcNow.AddMilliseconds(WaitMs); + while (!condition()) + { + if (DateTime.UtcNow > deadline) throw new ShouldAssertException(because); + await Task.Delay(10); + } + } + + private static HostState? StateOf(GalaxyDriver driver, string host) => + driver.GetHostStatuses().FirstOrDefault(s => s.HostName == host)?.State; + + /// + /// A ScanState event on the probe handle updates GetHostStatuses; after a reconnect replay + /// the probe is re-subscribed and events on the NEW handle keep updating it, while the dead + /// pre-reconnect handle no longer does. Probe traffic never reaches public OnDataChange. + /// + [Fact] + public async Task ScanState_event_updates_GetHostStatuses_and_survives_a_reconnect() + { + var sub = new StreamingSubscriber(); + var source = new CountingHierarchySource(); + await using var driver = new GalaxyDriver("drv", Opts(), hierarchySource: source, subscriber: sub); + var publicRefs = new List(); + driver.OnDataChange += (_, e) => { lock (publicRefs) publicRefs.Add(e.FullReference); }; + + await driver.InitializeAsync("{}", CancellationToken.None); + await driver.DiscoverAsync(new NullBuilder(), CancellationToken.None); + // The pump starts on the first value subscription (the probe watcher does not start it). + await driver.SubscribeAsync(["Tank01.Level"], TimeSpan.Zero, CancellationToken.None); + + var probeHandle = sub.HandleByRef["Plat01.ScanState"]; + await sub.EmitBoolAsync(probeHandle, true); + await WaitUntilAsync(() => StateOf(driver, "Plat01") == HostState.Running, + "a ScanState=true event must surface Plat01 as Running"); + + // Reconnect replay: every pre-reconnect handle is dead; the probe must rebind. + await driver.InvokeReplayForTestAsync(CancellationToken.None); + var rebound = sub.HandleByRef["Plat01.ScanState"]; + rebound.ShouldNotBe(probeHandle, "replay must re-subscribe the probe and record the fresh handle"); + + await sub.EmitBoolAsync(rebound, false); + await WaitUntilAsync(() => StateOf(driver, "Plat01") == HostState.Stopped, + "a ScanState=false event on the post-reconnect handle must surface Plat01 as Stopped"); + + // The dead handle is no longer routed (positive control: the live one still is). + await sub.EmitBoolAsync(probeHandle, true); + await sub.EmitDoubleAsync(sub.HandleByRef["Tank01.Level"], 1.0); + await WaitUntilAsync(() => { lock (publicRefs) return publicRefs.Contains("Tank01.Level"); }, "value traffic must still flow"); + StateOf(driver, "Plat01").ShouldBe(HostState.Stopped, "an event on the pre-reconnect handle must be dropped"); + + lock (publicRefs) publicRefs.ShouldNotContain("Plat01.ScanState", "probe traffic is the watcher's, never the host's"); + } + + /// Discovery asks the gateway for the hierarchy once; the probe set is selected from that same sweep. + [Fact] + public async Task DiscoverAsync_fetches_the_hierarchy_once_and_still_syncs_the_probe_set() + { + var sub = new StreamingSubscriber(); + var source = new CountingHierarchySource(); + await using var driver = new GalaxyDriver("drv", Opts(), hierarchySource: source, subscriber: sub); + await driver.InitializeAsync("{}", CancellationToken.None); + + await driver.DiscoverAsync(new NullBuilder(), CancellationToken.None); + + source.Calls.ShouldBe(1, "the probe selection must reuse the discoverer's sweep"); + sub.Subscribed.ShouldContain("Plat01.ScanState", "positive control: the platform probe was still synced"); + driver.ProbeWatcherForTests.ShouldNotBeNull().WatchedPlatforms.ShouldBe(["Plat01"]); + } + + /// Inbound stream data advances LastSuccessfulRead (throttled); the throttle holds the stamp still. + [Fact] + public async Task Stream_data_advances_LastSuccessfulRead_under_a_throttle() + { + var sub = new StreamingSubscriber(); + await using var driver = new GalaxyDriver("drv", Opts(), hierarchySource: null, subscriber: sub) + { + LivenessStampThrottle = TimeSpan.FromHours(1), + }; + await driver.InitializeAsync("{}", CancellationToken.None); + var initStamp = driver.GetHealth().LastSuccessfulRead.ShouldNotBeNull(); + await Task.Delay(20); + + var delivered = 0; + driver.OnDataChange += (_, _) => Interlocked.Increment(ref delivered); + await driver.SubscribeAsync(["Tank01.Level"], TimeSpan.Zero, CancellationToken.None); + var handle = sub.HandleByRef["Tank01.Level"]; + + await sub.EmitDoubleAsync(handle, 1.0); + await WaitUntilAsync(() => Volatile.Read(ref delivered) == 1, "first event must be delivered"); + var first = driver.GetHealth().LastSuccessfulRead.ShouldNotBeNull(); + first.ShouldBeGreaterThan(initStamp, "data on the stream is liveness evidence"); + + await Task.Delay(20); + await sub.EmitDoubleAsync(handle, 2.0); + await WaitUntilAsync(() => Volatile.Read(ref delivered) == 2, "second event must be delivered"); + driver.GetHealth().LastSuccessfulRead.ShouldBe(first, "inside the throttle window the stamp holds still"); + } +}