diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Messages/Drivers/DriverHealthChanged.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Messages/Drivers/DriverHealthChanged.cs index b269531d..88da2aaa 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Messages/Drivers/DriverHealthChanged.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Messages/Drivers/DriverHealthChanged.cs @@ -13,6 +13,19 @@ namespace ZB.MOM.WW.OtOpcUa.Commons.Messages.Drivers; /// Latest error message; null when none. /// Number of state-transitions into Faulted in the last 5 minutes. /// Timestamp this snapshot was published. +/// +/// When the driver last raised IRediscoverable.OnRediscoveryNeeded — i.e. it observed that the +/// remote's tag set may have changed (a Galaxy redeploy, a TwinCAT symbol-version bump, an MTConnect +/// agent restart, a Sparkplug rebirth). Null when the driver has never raised it, or does not +/// implement IRediscoverable. +/// Advisory only. The served address space is NOT rebuilt in response — raw tags are +/// authored explicitly through the /raw browse-commit flow. This is a prompt for an operator +/// to re-browse the device and commit whatever changed. +/// +/// +/// The driver-supplied reason string from the same event (e.g. "deploy-time-changed"), shown +/// to the operator alongside the prompt. Null when is null. +/// public sealed record DriverHealthChanged( string ClusterId, string DriverInstanceId, @@ -20,7 +33,9 @@ public sealed record DriverHealthChanged( DateTime? LastSuccessfulReadUtc, string? LastError, int ErrorCount5Min, - DateTime PublishedUtc) + DateTime PublishedUtc, + DateTime? RediscoveryNeededUtc = null, + string? RediscoveryReason = null) { /// /// DPS topic name. Both the runtime AkkaDriverHealthPublisher and the AdminUI diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Protos/telemetry.proto b/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Protos/telemetry.proto index 0997fb2e..f619d29b 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Protos/telemetry.proto +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Protos/telemetry.proto @@ -68,6 +68,8 @@ message DriverHealth { optional string last_error = 5; // nullable in the record int32 error_count_5min = 6; google.protobuf.Timestamp published_utc = 7; + google.protobuf.Timestamp rediscovery_needed_utc = 8; // DateTime? — absent Timestamp encodes null + optional string rediscovery_reason = 9; // nullable in the record } // Mirrors ZB.MOM.WW.OtOpcUa.Commons.Messages.Drivers.DriverResilienceStatusChanged. diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/IDriverHealthPublisher.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/IDriverHealthPublisher.cs index 32851fcb..8fff62c2 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/IDriverHealthPublisher.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/IDriverHealthPublisher.cs @@ -15,11 +15,20 @@ public interface IDriverHealthPublisher /// The driver instance the snapshot describes. /// The current health snapshot. /// The number of errors observed in the trailing 5-minute window. + /// + /// When the driver last raised , or null if never + /// (or if it is not an ). Advisory: it prompts an operator to + /// re-browse, and never rebuilds the served address space. + /// + /// The driver-supplied reason from that event; null when + /// is null. void Publish( string clusterId, string driverInstanceId, DriverHealth health, - int errorCount5Min); + int errorCount5Min, + DateTime? rediscoveryNeededUtc = null, + string? rediscoveryReason = null); } /// @@ -38,6 +47,8 @@ public sealed class NullDriverHealthPublisher : IDriverHealthPublisher string clusterId, string driverInstanceId, DriverHealth health, - int errorCount5Min) + int errorCount5Min, + DateTime? rediscoveryNeededUtc = null, + string? rediscoveryReason = null) { /* no-op */ } } diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Components/Pages/Hosts.razor b/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Components/Pages/Hosts.razor index 107b1bd6..f121a70c 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Components/Pages/Hosts.razor +++ b/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Components/Pages/Hosts.razor @@ -191,6 +191,16 @@ else { @d.DriverInstanceId } + @if (d.RediscoveryNeededUtc is not null) + { + @* The driver reported its remote's tag set may have changed. Advisory + only — v3 authors raw tags via /raw browse-commit, so nothing is + grafted at runtime and an operator must re-browse to pick it up. *@ + + re-browse + + } @(d.DriverType ?? "—") @d.State diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Hosts/HostsDriverView.cs b/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Hosts/HostsDriverView.cs index 385a6838..4940f313 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Hosts/HostsDriverView.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Hosts/HostsDriverView.cs @@ -35,9 +35,15 @@ public sealed record HostsDriverInstanceInfo(string DriverInstanceId, string Clu /// Latest error message; null when none. /// Faulted-transition count in the last 5 minutes. /// Timestamp the snapshot was published. +/// When the driver last reported that the remote's tag set may have +/// changed; null when never (or when the driver cannot report it). Advisory — the served address space +/// is unchanged and an operator must re-browse the device via /raw to pick anything up. +/// The driver-supplied reason for that report; null when +/// is null. public sealed record HostsDriverRow( string DriverInstanceId, string? Name, string? DriverType, string State, - DateTime? LastSuccessfulReadUtc, string? LastError, int ErrorCount5Min, DateTime PublishedUtc); + DateTime? LastSuccessfulReadUtc, string? LastError, int ErrorCount5Min, DateTime PublishedUtc, + DateTime? RediscoveryNeededUtc = null, string? RediscoveryReason = null); /// /// One cluster's section on the /hosts page: its configured nodes plus its enriched @@ -110,7 +116,9 @@ public static class HostsDriverView s.LastSuccessfulReadUtc, s.LastError, s.ErrorCount5Min, - s.PublishedUtc); + s.PublishedUtc, + s.RediscoveryNeededUtc, + s.RediscoveryReason); }) .OrderBy(d => d.Name ?? d.DriverInstanceId, StringComparer.OrdinalIgnoreCase) .ThenBy(d => d.DriverInstanceId, StringComparer.OrdinalIgnoreCase) diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Telemetry/TelemetryProtoMapCentral.cs b/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Telemetry/TelemetryProtoMapCentral.cs index f851f761..bae54518 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Telemetry/TelemetryProtoMapCentral.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Telemetry/TelemetryProtoMapCentral.cs @@ -118,7 +118,9 @@ public static class TelemetryProtoMapCentral LastSuccessfulReadUtc: msg.LastSuccessfulReadUtc?.ToDateTime(), LastError: msg.HasLastError ? msg.LastError : null, ErrorCount5Min: msg.ErrorCount5Min, - PublishedUtc: Required(msg.PublishedUtc, "DriverHealth", "published_utc")); + PublishedUtc: Required(msg.PublishedUtc, "DriverHealth", "published_utc"), + RediscoveryNeededUtc: msg.RediscoveryNeededUtc?.ToDateTime(), + RediscoveryReason: msg.HasRediscoveryReason ? msg.RediscoveryReason : null); } /// Projects a onto a . diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Grpc/TelemetryProtoMapNode.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Grpc/TelemetryProtoMapNode.cs index 8682a16f..066c6de1 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Grpc/TelemetryProtoMapNode.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Grpc/TelemetryProtoMapNode.cs @@ -128,6 +128,10 @@ public static class TelemetryProtoMapNode msg.LastSuccessfulReadUtc = ToUtcTimestamp(e.LastSuccessfulReadUtc.Value); if (e.LastError is not null) msg.LastError = e.LastError; + if (e.RediscoveryNeededUtc is not null) + msg.RediscoveryNeededUtc = ToUtcTimestamp(e.RediscoveryNeededUtc.Value); + if (e.RediscoveryReason is not null) + msg.RediscoveryReason = e.RediscoveryReason; return msg; } diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/AkkaDriverHealthPublisher.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/AkkaDriverHealthPublisher.cs index 6f93d39f..d53a4799 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/AkkaDriverHealthPublisher.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/AkkaDriverHealthPublisher.cs @@ -30,7 +30,13 @@ public sealed class AkkaDriverHealthPublisher : IDriverHealthPublisher } /// - public void Publish(string clusterId, string driverInstanceId, DriverHealth health, int errorCount5Min) + public void Publish( + string clusterId, + string driverInstanceId, + DriverHealth health, + int errorCount5Min, + DateTime? rediscoveryNeededUtc = null, + string? rediscoveryReason = null) { var msg = new DriverHealthChanged( clusterId, @@ -39,7 +45,9 @@ public sealed class AkkaDriverHealthPublisher : IDriverHealthPublisher health.LastSuccessfulRead, health.LastError, errorCount5Min, - DateTime.UtcNow); + DateTime.UtcNow, + rediscoveryNeededUtc, + rediscoveryReason); DistributedPubSub.Get(_system).Mediator.Tell(new Publish(TopicName, msg)); // Phase 5: fan the same snapshot into the node-local live-telemetry hub (no-op until a gRPC // client subscribes). The DPS publish above is unchanged — the hub is a strictly additive tap. diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverInstanceActor.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverInstanceActor.cs index c0ca7f97..b8a24c08 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverInstanceActor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverInstanceActor.cs @@ -139,6 +139,13 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers /// public sealed record TriggerRediscovery; + /// Self-sent when the wrapped driver raises — + /// it observed that the remote's tag set may have changed. Marshals the event off the driver's thread + /// onto the actor thread. Handled in EVERY behaviour, including Stubbed and Reconnecting: the raise has no + /// connection affinity (a Galaxy redeploy or a TwinCAT symbol-version bump can land while the driver is + /// between connects), and dropping it in one state would lose the signal silently. + private sealed record RediscoveryRaised(RediscoveryEventArgs Args); + /// Internal self-tick driving bounded post-connect re-discovery (FixedTree populates ~0–2s after connect). /// is the ordered-distinct full-reference signature of the prior pass's /// captured set (empty string on the first tick); re-discovery stops once a non-empty set repeats it. @@ -203,6 +210,16 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers private ISubscriptionHandle? _subscriptionHandle; private EventHandler? _dataChangeHandler; private EventHandler? _alarmEventHandler; + private EventHandler? _rediscoveryHandler; + + /// When the driver last raised , and the reason + /// it gave. Null until the first raise. Carried on every subsequent health snapshot so the AdminUI can + /// prompt an operator to re-browse the device. + /// Deliberately sticky — it is not cleared on reconnect or on a later clean pass. The remote's + /// tag set stayed changed; only an operator re-browsing and committing resolves it, and this actor cannot + /// observe that happening. A redeploy respawns the child, which clears it naturally. + private DateTime? _rediscoveryNeededUtc; + private string? _rediscoveryReason; /// The references the host wants kept subscribed (set by ). /// Re-applied on every entry into Connected so values resume after a reconnect or redeploy. @@ -344,6 +361,9 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers // Warm up the snapshot store immediately so AdminUI sees current state as soon as the // actor starts, before any state transition fires. Also start the periodic heartbeat so // long-lived Healthy drivers keep their snapshot fresh for newly-joined SignalR clients. + // Attach the rediscovery signal before the first publish. Not per-connect: an IRediscoverable raise + // has no connection affinity, and a driver can observe a remote change while disconnected. + AttachRediscoverySource(); PublishHealthSnapshot(); Timers.StartPeriodicTimer("health-poll", HealthPollTick.Instance, _healthPollInterval); } @@ -365,6 +385,7 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers Receive(_ => { }); // A TriggerRediscovery is meaningless to a stubbed (never-Connected) driver — silently ignore it. Receive(_ => { }); + Receive(HandleRediscoveryRaised); Receive(_ => PublishHealthSnapshot()); } @@ -424,6 +445,7 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers // A TriggerRediscovery arriving while not Connected is a deliberate no-op — the (re)connect path // re-runs discovery anyway. Swallow it so it stays a clean silent no-op (no Unhandled event). Receive(_ => { }); + Receive(HandleRediscoveryRaised); Receive(_ => PublishHealthSnapshot()); } @@ -499,6 +521,7 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers // to Self (HandleSubscribeAsync already logged the underlying cause). Swallow it so it doesn't dead-letter. Receive(msg => _log.Debug("DriverInstance {Id}: resubscribe reported failure: {Reason}", _driverInstanceId, msg.Reason)); + Receive(HandleRediscoveryRaised); Receive(_ => { PublishHealthSnapshot(); @@ -608,6 +631,7 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers // A TriggerRediscovery arriving while not Connected is a deliberate no-op — the (re)connect path // re-runs discovery anyway. Swallow it so it stays a clean silent no-op (no Unhandled event). Receive(_ => { }); + Receive(HandleRediscoveryRaised); Receive(_ => PublishHealthSnapshot()); Timers.StartPeriodicTimer("retry-connect", RetryConnect.Instance, _reconnectInterval); } @@ -853,6 +877,44 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers src.OnAlarmEvent += _alarmEventHandler; } + /// Subscribe the driver's (if it is one), + /// marshaling each raise to the actor thread. Idempotent; mirrors . + /// Attached once in PreStart rather than per-connect, because the interesting raises + /// (a Galaxy redeploy, a TwinCAT symbol-version bump) can happen while the driver is between + /// connects, and the event carries no connection affinity. + private void AttachRediscoverySource() + { + if (_driver is not IRediscoverable src || _rediscoveryHandler is not null) return; + var self = Self; + _rediscoveryHandler = (_, e) => self.Tell(new RediscoveryRaised(e)); + src.OnRediscoveryNeeded += _rediscoveryHandler; + } + + /// Symmetric teardown, called from PostStop. Load-bearing: the instance + /// can OUTLIVE this actor (the host respawns a child around the same driver object), so a missing + /// unsubscribe would accumulate one handler per respawn, each holding a dead Self. + private void DetachRediscoverySource() + { + if (_driver is IRediscoverable src && _rediscoveryHandler is not null) + src.OnRediscoveryNeeded -= _rediscoveryHandler; + _rediscoveryHandler = null; + } + + /// Records the driver's rediscovery raise and re-publishes health so the signal reaches the + /// AdminUI promptly rather than waiting for the next 30 s heartbeat. + /// Advisory only. The served address space is deliberately NOT rebuilt: v3 authors raw tags + /// explicitly through the /raw browse-commit flow, so a runtime graft would create nodes no + /// operator approved and that no deployment artifact records. This prompts a human to re-browse. + private void HandleRediscoveryRaised(RediscoveryRaised msg) + { + _rediscoveryNeededUtc = DateTime.UtcNow; + _rediscoveryReason = msg.Args.Reason; + _log.Info( + "DriverInstance {Id}: driver reports its tag set may have changed ({Reason}) — surfaced for operator re-browse; the served address space is unchanged", + _driverInstanceId, msg.Args.Reason); + PublishHealthSnapshot(); + } + /// Symmetric teardown — called from and PostStop so a stale /// handler never pushes to a disconnected actor. private void DetachAlarmSource() @@ -1081,11 +1143,16 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers { var health = _driver.GetHealth(); var errorCount = ErrorCount5Min(); - var fingerprint = (health.State, health.LastSuccessfulRead, health.LastError, errorCount); + // _rediscoveryNeededUtc is PART OF THE FINGERPRINT on purpose. A rediscovery raise on an + // otherwise-unchanged Healthy driver leaves (state, lastSuccess, lastError, errorCount) + // identical, so without it the dedup below would swallow the very publish that carries the + // signal and the operator would never see the prompt. + var fingerprint = (health.State, health.LastSuccessfulRead, health.LastError, errorCount, _rediscoveryNeededUtc); if (_lastPublishedFingerprint is { } prev && prev.Equals(fingerprint)) return; _lastPublishedFingerprint = fingerprint; - _healthPublisher.Publish(_clusterId, _driverInstanceId, health, errorCount); + _healthPublisher.Publish( + _clusterId, _driverInstanceId, health, errorCount, _rediscoveryNeededUtc, _rediscoveryReason); } catch (Exception ex) { @@ -1094,12 +1161,15 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers } /// Fingerprint of the last call; null until first publish. - private (DriverState State, DateTime? LastSuccess, string? LastError, int ErrorCount)? _lastPublishedFingerprint; + private (DriverState State, DateTime? LastSuccess, string? LastError, int ErrorCount, DateTime? RediscoveryNeededUtc)? _lastPublishedFingerprint; /// protected override void PostStop() { DetachSubscription(); + // MUST happen: the IDriver instance can outlive this actor (the host respawns a child around the + // same driver object), so a missing unsubscribe accumulates a handler per respawn holding a dead Self. + DetachRediscoverySource(); try { _driver.ShutdownAsync(CancellationToken.None).GetAwaiter().GetResult(); } catch (Exception ex) { _log.Warning(ex, "DriverInstance {Id}: ShutdownAsync threw on PostStop", _driverInstanceId); } OtOpcUaTelemetry.DriverInstanceLifecycle.Add(1, diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverInstanceActorRediscoverySignalTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverInstanceActorRediscoverySignalTests.cs new file mode 100644 index 00000000..4027bf68 --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverInstanceActorRediscoverySignalTests.cs @@ -0,0 +1,249 @@ +using Akka.Actor; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; +using ZB.MOM.WW.OtOpcUa.Runtime.Drivers; +using ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness; + +namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Drivers; + +/// +/// Covers the consumer (Gitea #518). Nine drivers raise +/// when they observe that a remote's tag set may have +/// changed — a Galaxy redeploy, a TwinCAT symbol-version bump, an MTConnect agent restart, a Sparkplug +/// rebirth. Before this wiring NOTHING in src/ subscribed, so every one of those raises went into +/// the void. +/// The signal is advisory. It does not rebuild the served address space: v3 authors raw tags +/// explicitly through the /raw browse-commit flow, so a runtime graft would create nodes no +/// operator approved. It rides the driver-health snapshot to the AdminUI as a re-browse prompt. +/// +[Trait("Category", "Unit")] +public sealed class DriverInstanceActorRediscoverySignalTests : RuntimeActorTestBase +{ + /// + /// The load-bearing case: a driver raises rediscovery while otherwise perfectly healthy, and the + /// signal reaches the health publisher carrying the driver's reason. + /// + [Fact] + public void Rediscovery_raise_reaches_the_health_publisher_with_its_reason() + { + var driver = new RediscoverableStubDriver(); + var publisher = new RecordingHealthPublisher(); + var parent = CreateTestProbe(); + parent.IgnoreMessages(_ => true); + var actor = parent.ChildActorOf(DriverInstanceActor.Props(driver, healthPublisher: publisher)); + + actor.Tell(new DriverInstanceActor.InitializeRequested("{}")); + AwaitAssert(() => publisher.Published.Count.ShouldBeGreaterThan(0), TimeSpan.FromSeconds(3)); + + driver.RaiseRediscovery("deploy-time-changed"); + + AwaitAssert( + () => + { + var withSignal = publisher.Published.LastOrDefault(p => p.RediscoveryNeededUtc is not null); + withSignal.ShouldNotBeNull(); + withSignal!.RediscoveryReason.ShouldBe("deploy-time-changed"); + }, + TimeSpan.FromSeconds(3)); + } + + /// + /// Regression guard for the dedup trap. PublishHealthSnapshot suppresses a publish whose + /// (state, lastSuccessfulRead, lastError, errorCount) tuple repeats. A rediscovery raise on an + /// otherwise-unchanged Healthy driver leaves all four identical, so unless the rediscovery timestamp + /// is part of the fingerprint the dedup swallows the very publish carrying the signal and the operator + /// never sees the prompt. + /// Positive control: the assertion is that the count STRICTLY INCREASES across the raise. An + /// "eventually non-null" assertion alone would pass on the warm-up publish and prove nothing. Revert + /// _rediscoveryNeededUtc out of the fingerprint tuple and this test must fail. + /// + [Fact] + public void Rediscovery_raise_is_not_swallowed_by_the_unchanged_health_dedup() + { + var driver = new RediscoverableStubDriver(); + var publisher = new RecordingHealthPublisher(); + var parent = CreateTestProbe(); + parent.IgnoreMessages(_ => true); + var actor = parent.ChildActorOf(DriverInstanceActor.Props(driver, healthPublisher: publisher)); + + actor.Tell(new DriverInstanceActor.InitializeRequested("{}")); + AwaitAssert(() => publisher.Published.Count.ShouldBeGreaterThan(0), TimeSpan.FromSeconds(3)); + + // Let the actor settle so nothing else is in flight, then capture the baseline. Everything about + // the driver's health is now stable — only the rediscovery flag will change. + ExpectNoMsg(TimeSpan.FromMilliseconds(200)); + var before = publisher.Published.Count; + + driver.RaiseRediscovery("symbol-version-changed"); + + AwaitAssert( + () => publisher.Published.Count.ShouldBeGreaterThan(before), + TimeSpan.FromSeconds(3)); + publisher.Published[^1].RediscoveryNeededUtc.ShouldNotBeNull(); + } + + /// + /// The signal is sticky: a later health publish still carries it. The remote's tag set stayed + /// changed, and only an operator re-browsing resolves that — which this actor cannot observe. A + /// flag that cleared itself on the next heartbeat would be a prompt the operator could miss entirely. + /// + [Fact] + public void Rediscovery_flag_persists_on_subsequent_health_publishes() + { + var driver = new RediscoverableStubDriver(); + var publisher = new RecordingHealthPublisher(); + var parent = CreateTestProbe(); + parent.IgnoreMessages(_ => true); + var actor = parent.ChildActorOf(DriverInstanceActor.Props(driver, healthPublisher: publisher)); + + actor.Tell(new DriverInstanceActor.InitializeRequested("{}")); + AwaitAssert(() => publisher.Published.Count.ShouldBeGreaterThan(0), TimeSpan.FromSeconds(3)); + driver.RaiseRediscovery("agent-instance-changed"); + AwaitAssert( + () => publisher.Published.Any(p => p.RediscoveryNeededUtc is not null), + TimeSpan.FromSeconds(3)); + + // Force a further publish by changing something else about the driver's health. + driver.SetLastError("transient read failure"); + AwaitAssert( + () => publisher.Published.Any(p => p.LastError == "transient read failure"), + TimeSpan.FromSeconds(3)); + + var latest = publisher.Published[^1]; + latest.RediscoveryNeededUtc.ShouldNotBeNull(); + latest.RediscoveryReason.ShouldBe("agent-instance-changed"); + } + + /// + /// Leak guard. The instance can OUTLIVE the actor — the host stops a + /// child and respawns another around the same driver object — so a missing -= in + /// PostStop accumulates one handler per respawn, each holding a dead Self. Asserts the + /// driver's invocation list is empty again after the actor stops. + /// + [Fact] + public void Stopping_the_actor_detaches_the_rediscovery_handler() + { + var driver = new RediscoverableStubDriver(); + var parent = CreateTestProbe(); + parent.IgnoreMessages(_ => true); + var actor = parent.ChildActorOf(DriverInstanceActor.Props(driver)); + + actor.Tell(new DriverInstanceActor.InitializeRequested("{}")); + AwaitAssert(() => driver.SubscriberCount.ShouldBe(1), TimeSpan.FromSeconds(3)); + + Watch(actor); + actor.Tell(PoisonPill.Instance); + ExpectTerminated(actor, TimeSpan.FromSeconds(3)); + + driver.SubscriberCount.ShouldBe(0); + } + + /// A driver that does NOT implement must publish a null flag — + /// the field is absent, not defaulted to "needs re-browse". + [Fact] + public void A_non_rediscoverable_driver_never_sets_the_flag() + { + var publisher = new RecordingHealthPublisher(); + var parent = CreateTestProbe(); + parent.IgnoreMessages(_ => true); + var actor = parent.ChildActorOf(DriverInstanceActor.Props(new StubDriver(), healthPublisher: publisher)); + + actor.Tell(new DriverInstanceActor.InitializeRequested("{}")); + AwaitAssert(() => publisher.Published.Count.ShouldBeGreaterThan(0), TimeSpan.FromSeconds(3)); + + publisher.Published.ShouldAllBe(p => p.RediscoveryNeededUtc == null); + } + + /// Captures every health publish so a test can assert on the rediscovery fields. + private sealed record HealthPublish( + string ClusterId, + string DriverInstanceId, + DriverHealth Health, + int ErrorCount5Min, + DateTime? RediscoveryNeededUtc, + string? RediscoveryReason) + { + public string? LastError => Health.LastError; + } + + private sealed class RecordingHealthPublisher : IDriverHealthPublisher + { + private readonly List _published = []; + + /// Thread-safe snapshot — Publish is called from the actor thread while the test + /// asserts from its own. + public IReadOnlyList Published + { + get { lock (_published) return _published.ToArray(); } + } + + /// + public void Publish( + string clusterId, + string driverInstanceId, + DriverHealth health, + int errorCount5Min, + DateTime? rediscoveryNeededUtc = null, + string? rediscoveryReason = null) + { + lock (_published) + { + _published.Add(new HealthPublish( + clusterId, driverInstanceId, health, errorCount5Min, rediscoveryNeededUtc, rediscoveryReason)); + } + } + } + + /// + /// A stub driver exposing , a hook to raise it, and the live subscriber + /// count (for the detach guard). + /// Implemented from scratch rather than derived from the shared StubDriver on + /// purpose. That one returns GetHealth() => new(Healthy, DateTime.UtcNow, null), so its + /// health fingerprint differs on EVERY call and the dedup under test never engages — which would make + /// pass vacuously + /// whether or not the fix is present. This stub's health is STABLE until a test changes it. + /// + private sealed class RediscoverableStubDriver : IDriver, IRediscoverable + { + private static readonly DateTime FixedLastRead = new(2026, 7, 27, 12, 0, 0, DateTimeKind.Utc); + private volatile string? _lastError; + + /// + public event EventHandler? OnRediscoveryNeeded; + + /// + public string DriverInstanceId => "rediscoverable-stub-1"; + + /// + public string DriverType => "Stub"; + + /// Number of live subscribers on . + public int SubscriberCount => OnRediscoveryNeeded?.GetInvocationList().Length ?? 0; + + /// Raises the event exactly as a real driver's watcher does. + public void RaiseRediscovery(string reason) + => OnRediscoveryNeeded?.Invoke(this, new RediscoveryEventArgs(reason, ScopeHint: null)); + + /// Changes the reported health so a test can force a publish that is NOT the rediscovery one. + public void SetLastError(string error) => _lastError = error; + + /// + public Task InitializeAsync(string driverConfigJson, CancellationToken cancellationToken) => Task.CompletedTask; + + /// + public Task ReinitializeAsync(string driverConfigJson, CancellationToken cancellationToken) => Task.CompletedTask; + + /// + public Task ShutdownAsync(CancellationToken cancellationToken) => Task.CompletedTask; + + /// + public DriverHealth GetHealth() => new(DriverState.Healthy, FixedLastRead, _lastError); + + /// + public long GetMemoryFootprint() => 0; + + /// + public Task FlushOptionalCachesAsync(CancellationToken cancellationToken) => Task.CompletedTask; + } +} diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverInstanceActorSubscriptionReconcileTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverInstanceActorSubscriptionReconcileTests.cs index 1a901650..d331ed3a 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverInstanceActorSubscriptionReconcileTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverInstanceActorSubscriptionReconcileTests.cs @@ -116,7 +116,13 @@ public sealed class DriverInstanceActorSubscriptionReconcileTests : RuntimeActor public int Count => Volatile.Read(ref _count); /// - public void Publish(string clusterId, string driverInstanceId, DriverHealth health, int errorCount5Min) + public void Publish( + string clusterId, + string driverInstanceId, + DriverHealth health, + int errorCount5Min, + DateTime? rediscoveryNeededUtc = null, + string? rediscoveryReason = null) => Interlocked.Increment(ref _count); } }