diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs index 31f13546..b751214f 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs @@ -205,6 +205,17 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS /// Node-wide script run-guard options for the scripted-alarm engine; null ⇒ compiled defaults. private readonly ScriptingOptions? _scripting; + + /// + /// Driver children this host has stopped (, an operator Restart) and is watching + /// until they terminate. A value burst from one of them is DROPPED in : + /// Context.Stop is asynchronous, so the old child keeps publishing for the rest of the apply + /// handler and its own stop, and forwarding those bursts after would + /// re-retain a removed driver's last Good in the dependency mux indefinitely. Akka delivers + /// after the last message the watched child sent here, so removing the entry + /// there bounds the set without ever dropping a live incarnation's values. + /// + private readonly HashSet _retiredChildren = new(); private readonly IActorRef? _scriptedAlarmHostOverride; private readonly ILoggingAdapter _log = Context.GetLogger(); @@ -1516,6 +1527,7 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS Receive(HandleApplyWedged); Receive(HandleGetDiagnostics); Receive(ForwardBatchToMux); + Receive(OnRetiredChildTerminated); Receive(ForwardNativeAlarm); Receive(OnDriverConnectivityChanged); Receive(HandleCommsRecoveryAudit); @@ -1578,6 +1590,7 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS Receive(HandleApplyWedged); Receive(HandleGetDiagnostics); Receive(ForwardBatchToMux); + Receive(OnRetiredChildTerminated); Receive(ForwardNativeAlarm); Receive(OnDriverConnectivityChanged); Receive(HandleCommsRecoveryAudit); @@ -1622,6 +1635,16 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS /// private void ForwardBatchToMux(DriverInstanceActor.AttributeValueBatchPublished msg) { + // A burst from an incarnation this host already stopped: forwarding it would undo the SourceRemoved + // eviction (see _retiredChildren). Its ConnectivityChanged is NOT filtered — the respawn path relies on + // that Bad fan to cover the gap. + if (_retiredChildren.Contains(Sender)) + { + _log.Debug("DriverHost {Node}: dropped a {Count}-item value burst from a stopped incarnation of {Id}", + _localNode, msg.Items.Count, msg.DriverInstanceId); + return; + } + // One envelope per burst; the mux filters per item (it registered a batch receive). _dependencyMux?.Tell(msg); @@ -2517,6 +2540,7 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS Receive(m => _log.Debug("DriverHost {Node}: dropping a {Count}-item value burst from {Id} while Stale (no address space)", _localNode, m.Items.Count, m.DriverInstanceId)); + Receive(OnRetiredChildTerminated); Receive(m => _log.Debug("DriverHost {Node}: dropping a native alarm transition ({Cond}) from {Id} while Stale", _localNode, m.Args.ConditionId, m.DriverInstanceId)); @@ -3832,6 +3856,7 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS { if (!_children.TryGetValue(driverInstanceId, out var entry)) return; StopDependencyAdapter(driverInstanceId); + RetireChild(entry.Actor); Context.Stop(entry.Actor); _children.Remove(driverInstanceId); EvictRetainedValues(driverInstanceId); @@ -3863,15 +3888,28 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS /// values travel, so the eviction is ordered ahead of them. /// /// - /// Residual: a burst the old child enqueued on this host just before it stopped can still be forwarded - /// after the eviction and re-retain a value. The window is one mailbox drain; a comms-loss fan or the - /// next real publish overwrites it, and the mux's grace sweep bounds it once no replay subscriber holds - /// the ref. + /// The stopped child keeps publishing until it actually terminates, and for a REMOVED driver nothing + /// would ever overwrite a value re-retained after this eviction (it never publishes again, and its + /// PostStop Bad fan resolves no refs once the apply has rebuilt the index without it). So the caller + /// also s it, and drops every burst from a + /// retired child. /// /// The driver instance that ended. private void EvictRetainedValues(string driverInstanceId) => _dependencyMux?.Tell(new DependencyMuxActor.SourceRemoved(driverInstanceId)); + /// Marks a child this host is about to stop as retired and watches it until it terminates. + /// The child being stopped. + private void RetireChild(IActorRef child) + { + _retiredChildren.Add(child); + Context.Watch(child); + } + + /// A retired child has terminated; nothing more can arrive from it. + /// The termination notice. + private void OnRetiredChildTerminated(Terminated msg) => _retiredChildren.Remove(msg.ActorRef); + private static string ActorNameFor(string driverInstanceId, long generation) { // Akka actor names cannot contain '/', ':', or whitespace. Mangle defensively. The monotonic @@ -3917,6 +3955,7 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS // Stop the existing child actor — DriverInstanceActor.PostStop calls ShutdownAsync. StopDependencyAdapter(msg.DriverInstanceId); + RetireChild(entry.Actor); Context.Stop(entry.Actor); _children.Remove(msg.DriverInstanceId); // An operator Restart builds a NEW driver object, so the old one's retained values are evicted like a diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorSourceRemovedTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorSourceRemovedTests.cs index a7a49f7a..0e7e48ac 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorSourceRemovedTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorSourceRemovedTests.cs @@ -1,9 +1,13 @@ using Akka.Actor; using Akka.TestKit; +using Microsoft.EntityFrameworkCore; using Shouldly; using Xunit; using ZB.MOM.WW.OtOpcUa.Commons.Engines; +using ZB.MOM.WW.OtOpcUa.Commons.Messages.Admin; +using ZB.MOM.WW.OtOpcUa.Commons.OpcUa; using ZB.MOM.WW.OtOpcUa.Commons.Types; +using ZB.MOM.WW.OtOpcUa.Configuration; using ZB.MOM.WW.OtOpcUa.Runtime.Drivers; using ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness; using ZB.MOM.WW.OtOpcUa.Runtime.VirtualTags; @@ -14,8 +18,10 @@ namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Drivers; /// The dependency mux retains a ref's last value for replay-opting subscribers and evicts a driver's values on /// . Before this, nothing in production sent that message, so a /// driver removed by a deploy kept its last Good in the mux and replayed it to the next replay-opting -/// subscriber as if it were current. DriverHostActor.StopChild now sends it. A supervision restart -/// re-uses the same driver and must NOT send it. All through a REAL mux and real driver children. +/// subscriber as if it were current. The host now sends it wherever a driver OBJECT ends (deploy removal, +/// config-change respawn, operator Restart), never on a supervision restart, and drops value bursts from the +/// child incarnations it has already stopped so a late burst cannot re-retain what was evicted. All through a +/// REAL mux and real driver children. /// public sealed class DriverHostActorSourceRemovedTests : RuntimeActorTestBase { @@ -32,88 +38,187 @@ public sealed class DriverHostActorSourceRemovedTests : RuntimeActorTestBase ("DI-2", "Modbus2", "dev2", "flow", "Double", false), ]; + private IActorRef _host = ActorRefs.Nobody; + private IActorRef _mux = ActorRefs.Nobody; + private TestProbe _coordinator = null!; + private TestProbe _holder = null!; + private RecordingSubscribableDriverFactory _factory = null!; + private IDbContextFactory _db = null!; + /// Spawns a host over a real mux with two enabled drivers, waits for both children to subscribe, /// registers a replay-opting holder (so the mux retains both refs) and publishes one value per driver. - private (IActorRef Host, TestProbe Coordinator, RecordingSubscribableDriverFactory Factory) ArrangeTwoDriversWithRetainedValues( - Akka.TestKit.TestProbe holder) + private void ArrangeTwoDriversWithRetainedValues() { - var db = NewInMemoryDbFactory(); - _db = db; - var revA = V3ArtifactSeeder.SeedV3Deployment(db, RevA, BothDrivers, NoRefs, enabled: true); - var factory = new RecordingSubscribableDriverFactory("Modbus"); - var mux = Sys.ActorOf(DependencyMuxActor.Props()); - var coordinator = CreateTestProbe(); - var host = Sys.ActorOf(DriverHostActor.Props( - db, TestNode, coordinator.Ref, - driverFactory: factory, + _db = NewInMemoryDbFactory(); + var revA = V3ArtifactSeeder.SeedV3Deployment(_db, RevA, BothDrivers, NoRefs, enabled: true); + _factory = new RecordingSubscribableDriverFactory("Modbus"); + _mux = Sys.ActorOf(DependencyMuxActor.Props()); + _coordinator = CreateTestProbe(); + _holder = CreateTestProbe(); + _host = Sys.ActorOf(DriverHostActor.Props( + _db, TestNode, _coordinator.Ref, + driverFactory: _factory, localRoles: new HashSet { "driver" }, - dependencyMux: mux, + dependencyMux: _mux, virtualTagEvaluator: NullVirtualTagEvaluator.Instance, virtualTagHostOverride: CreateTestProbe().Ref)); - V3ArtifactSeeder.Dispatch(host, coordinator, revA, RevA); - AwaitAssert(() => factory.Created.Count.ShouldBe(2), PresenceBudget); - foreach (var d in factory.Created) AwaitAssert(() => d.SubscribeCount.ShouldBeGreaterThanOrEqualTo(1), PresenceBudget); + V3ArtifactSeeder.Dispatch(_host, _coordinator, revA, RevA); + AwaitAssert(() => _factory.Created.Count.ShouldBe(2), PresenceBudget); + // The spawn hands a child the PREVIOUS apply's set (empty on the first deploy) and the bulk push then + // re-subscribes with the real one; a value fired before that lands is lost, so wait for the real set. + AwaitSubscribed("DI-1", Path1); + AwaitSubscribed("DI-2", Path2); - mux.Tell(new DependencyMuxActor.RegisterInterest(new[] { Path1, Path2 }, holder.Ref, ReplayRetained: true)); - Driver(factory, "DI-1").FireDataChange(Path1, 11.0); - Driver(factory, "DI-2").FireDataChange(Path2, 22.0); - var seen = new HashSet(); - while (seen.Count < 2) - foreach (var item in holder.ExpectMsg(PresenceBudget).Items) - seen.Add(item.TagId); - - _mux = mux; - return (host, coordinator, factory); + _mux.Tell(new DependencyMuxActor.RegisterInterest(new[] { Path1, Path2 }, _holder.Ref, ReplayRetained: true)); + Driver("DI-1").FireDataChange(Path1, 11.0); + Driver("DI-2").FireDataChange(Path2, 22.0); + // One fish for both: the two drivers' bursts arrive in either order, and fishing for one at a time would + // discard the other. + var pending = new HashSet(StringComparer.Ordinal) { Path1, Path2 }; + _holder.FishForMessage(b => + { + foreach (var i in b.Items) + if ((i.TagId == Path1 && Equals(i.Value, 11.0)) || (i.TagId == Path2 && Equals(i.Value, 22.0))) + pending.Remove(i.TagId); + return pending.Count == 0; + }, PresenceBudget); } - private IActorRef _mux = ActorRefs.Nobody; - private Microsoft.EntityFrameworkCore.IDbContextFactory _db = null!; + /// Waits until the holder has been delivered for . + private void AwaitHolder(string path, object value) => + _holder.FishForMessage( + b => b.Items.Any(i => i.TagId == path && Equals(i.Value, value)), PresenceBudget); - private static RecordingSubscribableDriver Driver(RecordingSubscribableDriverFactory f, string id) => - f.Created.Last(d => d.DriverInstanceId == id); + private void AwaitSubscribed(string driverId, string path) => + AwaitAssert(() => (Driver(driverId).LastSubscribedRefs ?? []).ShouldContain(path), PresenceBudget); - /// A fresh replay-opting subscriber for both refs; returns the refs it was replayed. - private string[] ReplayedRefs() + private RecordingSubscribableDriver Driver(string id) => _factory.Created.Last(d => d.DriverInstanceId == id); + + /// A fresh replay-opting subscriber for both refs; returns what it was replayed, keyed by ref. + private Dictionary Replayed() { var probe = CreateTestProbe(); _mux.Tell(new DependencyMuxActor.RegisterInterest(new[] { Path1, Path2 }, probe.Ref, ReplayRetained: true)); return probe.ExpectMsg(PresenceBudget) - .Items.Select(i => i.TagId).OrderBy(x => x, StringComparer.Ordinal).ToArray(); + .Items.ToDictionary(i => i.TagId, StringComparer.Ordinal); + } + + /// The live child actor for (named drv-{id}-g{generation}). + private IActorRef ChildOf(string driverId) + { + var probe = CreateTestProbe(); + Sys.ActorSelection(_host.Path.ToStringWithoutAddress() + "/drv-" + driverId + "-g*").Tell(new Identify(driverId), probe.Ref); + return probe.ExpectMsg(PresenceBudget).Subject.ShouldNotBeNull(); + } + + private static void ShouldNotBeTheOldGood(Dictionary replay, string path, object oldValue) + { + if (!replay.TryGetValue(path, out var item)) return; + (Equals(item.Value, oldValue) && OpcUaStatus.IsGood(item.StatusCode)) + .ShouldBeFalse($"the stopped incarnation's last Good for {path} was replayed as current"); } [Fact] public void A_driver_removed_by_a_deploy_is_no_longer_replayed_while_a_still_present_driver_is() { - var holder = CreateTestProbe(); - var (host, coordinator, _) = ArrangeTwoDriversWithRetainedValues(holder); + ArrangeTwoDriversWithRetainedValues(); // Positive control before the removal: both values replay. - ReplayedRefs().ShouldBe(new[] { Path2, Path1 }.OrderBy(x => x, StringComparer.Ordinal).ToArray()); + Replayed().Keys.ShouldBe(new[] { Path1, Path2 }, ignoreOrder: true); // Deploy B drops DI-1 (StopChild). - var revB = V3ArtifactSeeder.SeedV3Deployment( - _db, RevB, [("DI-2", "Modbus2", "dev2", "flow", "Double", false)], NoRefs, enabled: true); - V3ArtifactSeeder.Dispatch(host, coordinator, revB, RevB); + var revB = V3ArtifactSeeder.SeedV3Deployment(_db, RevB, [("DI-2", "Modbus2", "dev2", "flow", "Double", false)], NoRefs, enabled: true); + V3ArtifactSeeder.Dispatch(_host, _coordinator, revB, RevB); AwaitAssert( - () => ReplayedRefs().ShouldBe(new[] { Path2 }, "DI-1 was removed: its last Good must not replay; DI-2's must"), + () => Replayed().Keys.ToArray().ShouldBe(new[] { Path2 }, "DI-1 was removed: its last Good must not replay; DI-2's must"), PresenceBudget); } + /// Review I-1: a burst the removed child sent after the eviction (it stops asynchronously, so it keeps + /// publishing through the rest of the apply) must be dropped by the host, not forwarded to re-retain the value. + [Fact] + public void A_late_burst_from_a_removed_child_does_not_re_retain_its_value() + { + ArrangeTwoDriversWithRetainedValues(); + var oldChild = ChildOf("DI-1"); + // Hold the removed child in its PostStop so it has NOT terminated when the stale burst lands; a terminated + // actor cannot send, and the host forgets a retired child on Terminated. + var gate = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + Driver("DI-1").ShutdownGate = gate; + + var revB = V3ArtifactSeeder.SeedV3Deployment(_db, RevB, [("DI-2", "Modbus2", "dev2", "flow", "Double", false)], NoRefs, enabled: true); + V3ArtifactSeeder.Dispatch(_host, _coordinator, revB, RevB); + + // The stale burst, sent AS the stopped child. + _host.Tell(ValuePublishes.Single("DI-1", Path1, 11.0, OpcUaStatus.Good, DateTime.UtcNow), oldChild); + // Positive control / ordering fence: a live driver's value sent after it is delivered, so the stale burst + // has been processed by the host by the time we look. + Driver("DI-2").FireDataChange(Path2, 23.0); + AwaitHolder(Path2, 23.0); + + Replayed().Keys.ToArray().ShouldBe(new[] { Path2 }, "the removed child's late burst re-retained its last Good"); + gate.TrySetResult(); + } + + /// Review m-3: a config edit respawns the driver (stop + spawn); the OLD object's value is evicted, and + /// the NEW driver's value replays (positive control). + [Fact] + public void A_config_change_respawn_evicts_the_old_value_and_the_new_drivers_value_replays() + { + ArrangeTwoDriversWithRetainedValues(); + + // Adding a tag to DI-1 changes its merged DriverConfig, which respawns it. + var revB = V3ArtifactSeeder.SeedV3Deployment(_db, RevB, + [.. BothDrivers, ("DI-1", "Modbus", "dev1", "temp", "Double", false)], NoRefs, enabled: true); + V3ArtifactSeeder.Dispatch(_host, _coordinator, revB, RevB); + AwaitAssert(() => _factory.Created.Count.ShouldBe(3), PresenceBudget); + var respawned = Driver("DI-1"); + // The spawn hands the child the PREVIOUS apply's set, then the bulk push re-subscribes with the new one + // (which names the added tag); a value fired between the two can be lost, so wait for the final set. + AwaitAssert(() => (respawned.LastSubscribedRefs ?? []).ShouldContain("Plant/Modbus/dev1/temp"), PresenceBudget); + + ShouldNotBeTheOldGood(Replayed(), Path1, 11.0); + + respawned.FireDataChange(Path1, 33.0); + AwaitHolder(Path1, 33.0); + var replay = Replayed(); + replay[Path1].Value.ShouldBe(33.0, "the respawned driver's own value replays"); + replay[Path2].Value.ShouldBe(22.0, "the untouched driver is unaffected"); + } + + /// Review m-3: an operator Restart builds a new driver object; the old one's value is evicted and the + /// new one's replays. + [Fact] + public void An_operator_restart_evicts_the_old_value_and_the_new_drivers_value_replays() + { + ArrangeTwoDriversWithRetainedValues(); + + _host.Tell(new RestartDriver("c1", "DI-1", "tester", Guid.NewGuid())); + AwaitAssert(() => _factory.Created.Count.ShouldBe(3), PresenceBudget); + var restarted = Driver("DI-1"); + AwaitSubscribed("DI-1", Path1); + + ShouldNotBeTheOldGood(Replayed(), Path1, 11.0); + + restarted.FireDataChange(Path1, 44.0); + AwaitHolder(Path1, 44.0); + Replayed()[Path1].Value.ShouldBe(44.0); + } + [Fact] public void A_supervision_restart_does_not_evict_the_drivers_retained_value() { - var holder = CreateTestProbe(); - var (_, _, factory) = ArrangeTwoDriversWithRetainedValues(holder); - var di1 = Driver(factory, "DI-1"); + ArrangeTwoDriversWithRetainedValues(); + var di1 = Driver("DI-1"); // Crash DI-1's child: Akka restarts it around the SAME driver object, and the host reprovisions it. di1.FireMalformedDataChange(Path1); AwaitAssert(() => di1.InitializeCount.ShouldBe(2), PresenceBudget); - factory.Created.Count.ShouldBe(2, "a supervision restart re-uses the driver; no new one is created"); + _factory.Created.Count.ShouldBe(2, "a supervision restart re-uses the driver; no new one is created"); - ReplayedRefs().ShouldBe( - new[] { Path2, Path1 }.OrderBy(x => x, StringComparer.Ordinal).ToArray(), + var replay = Replayed(); + replay.Keys.ShouldBe(new[] { Path1, Path2 }, ignoreOrder: true, "a restart is the same driver instance: its last value stays replayable"); } } diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/StubDrivers.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/StubDrivers.cs index 627a1c38..ea0cfaef 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/StubDrivers.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/StubDrivers.cs @@ -279,11 +279,18 @@ internal sealed class RecordingSubscribableDriver : IDriver, ISubscribable, IAla /// public Task ReinitializeAsync(string driverConfigJson, CancellationToken cancellationToken) => Task.CompletedTask; + /// + /// When set, awaits it (the actor bounds the wait), which keeps a stopping child + /// ALIVE in its PostStop for as long as the test wants — the window in which a stopped child can still + /// have messages reaching its parent. + /// + public TaskCompletionSource? ShutdownGate { get; set; } + /// public Task ShutdownAsync(CancellationToken cancellationToken) { Interlocked.Increment(ref _shutdownCount); - return Task.CompletedTask; + return ShutdownGate?.Task ?? Task.CompletedTask; } ///