From 5de1d661bb01d10caf74475c85ff8aeb848aea72 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 22 Oct 2026 14:36:54 -0400 Subject: [PATCH] fix(runtime): evict a stopped driver's retained values from the dependency mux DriverHostActor now tells the mux SourceRemoved(driverInstanceId) wherever a driver object ends: StopChild (deploy removal and the stop half of a config-change respawn) and an operator Restart. It is not sent on a supervision restart, which re-uses the same driver. Before this nothing sent it, so a removed driver's last Good stayed in the mux and replayed to the next replay-opting subscriber as current. DriverHostActorSourceRemovedTests drives a real mux with two real driver children: after a deploy drops DI-1 only DI-2's value replays (positive control: both replay before the removal), and a supervision restart of DI-1 leaves its value replayable. Residual (documented on EvictRetainedValues): a burst the old child enqueued just before it stopped can still be forwarded after the eviction; the next publish, a comms-loss fan or the grace sweep bounds it. Refs #731 #734 Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT --- .../Drivers/DriverHostActor.cs | 23 ++++ .../VirtualTags/DependencyMuxActor.cs | 8 +- .../DriverHostActorSourceRemovedTests.cs | 119 ++++++++++++++++++ 3 files changed, 145 insertions(+), 5 deletions(-) create mode 100644 tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorSourceRemovedTests.cs 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 17a43f9d..b3339e42 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs @@ -3840,6 +3840,7 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS StopDependencyAdapter(driverInstanceId); Context.Stop(entry.Actor); _children.Remove(driverInstanceId); + EvictRetainedValues(driverInstanceId); // The instance left the plan (StopChild is only reached via plan.ToStop), so its cached // Health/Resilience snapshots must not replay to new telemetry subscribers. Release the shared // resilience state FIRST: the resilience publisher re-emits the tracker's snapshots every tick, so @@ -3858,6 +3859,25 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS _log.Info("DriverHost {Node}: stopped driver child {Id}", _localNode, driverInstanceId); } + /// + /// Tells the dependency mux to drop every retained value this driver instance produced, so a stopped + /// driver's last Good is never replayed to a later replay-opting subscriber as if it were current. Sent + /// wherever a driver OBJECT ends: a deploy removal, the stop half of a config-change respawn + /// () and an operator Restart (a new driver object). Deliberately NOT sent on a + /// supervision restart (): that re-uses the same driver, whose last value is + /// still its current one. Sent before any respawn, on the same host→mux pair the new incarnation's + /// 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 driver instance that ended. + private void EvictRetainedValues(string driverInstanceId) => + _dependencyMux?.Tell(new DependencyMuxActor.SourceRemoved(driverInstanceId)); + private static string ActorNameFor(string driverInstanceId, long generation) { // Akka actor names cannot contain '/', ':', or whitespace. Mangle defensively. The monotonic @@ -3905,6 +3925,9 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS StopDependencyAdapter(msg.DriverInstanceId); 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 + // removal's; the respawned driver republishes its own. + EvictRetainedValues(msg.DriverInstanceId); // The old incarnation's health must not read as the new one's until it publishes its own. _healthPublisher.Forget(msg.DriverInstanceId); diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/VirtualTags/DependencyMuxActor.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/VirtualTags/DependencyMuxActor.cs index 40e9bf19..3dd8441e 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/VirtualTags/DependencyMuxActor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/VirtualTags/DependencyMuxActor.cs @@ -52,11 +52,9 @@ namespace ZB.MOM.WW.OtOpcUa.Runtime.VirtualTags; /// Source removal. The mux records which driver instance last published each retained ref; /// evicts every value that driver produced, so a removed driver's last /// Good is never replayed as current. A comms-loss fan (a Bad batch) already overwrites the retained -/// value with Bad. ⚠️ As of fix round 1 no production actor sends yet — -/// DriverHostActor (outside the scripting lane) must Tell it when it stops / removes a driver -/// child. Until then a removed driver's last value can still be replayed to a replay-opting -/// subscriber that re-registers for that ref while another replay-opting subscriber holds it, or -/// within the grace window. +/// value with Bad. DriverHostActor sends wherever a driver object +/// ends (a deploy removal, the stop half of a config-change respawn, an operator Restart), and never +/// on a supervision restart, which re-uses the same driver. /// /// public sealed class DependencyMuxActor : ReceiveActor, IWithTimers 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 new file mode 100644 index 00000000..a7a49f7a --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorSourceRemovedTests.cs @@ -0,0 +1,119 @@ +using Akka.Actor; +using Akka.TestKit; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Commons.Engines; +using ZB.MOM.WW.OtOpcUa.Commons.Types; +using ZB.MOM.WW.OtOpcUa.Runtime.Drivers; +using ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness; +using ZB.MOM.WW.OtOpcUa.Runtime.VirtualTags; + +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. +/// +public sealed class DriverHostActorSourceRemovedTests : RuntimeActorTestBase +{ + private static readonly NodeId TestNode = NodeId.Parse("driver-sourceremoved-test"); + private static readonly RevisionHash RevA = RevisionHash.Parse(new string('a', 64)); + private static readonly RevisionHash RevB = RevisionHash.Parse(new string('b', 64)); + private const string Path1 = "Plant/Modbus/dev1/speed"; + private const string Path2 = "Plant/Modbus2/dev2/flow"; + private static readonly (string, string, string, string, string?)[] NoRefs = []; + + private static readonly (string Driver, string DriverName, string Device, string Tag, string DataType, bool Writable)[] BothDrivers = + [ + ("DI-1", "Modbus", "dev1", "speed", "Double", false), + ("DI-2", "Modbus2", "dev2", "flow", "Double", false), + ]; + + /// 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) + { + 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, + localRoles: new HashSet { "driver" }, + 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); + + 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); + } + + private IActorRef _mux = ActorRefs.Nobody; + private Microsoft.EntityFrameworkCore.IDbContextFactory _db = null!; + + private static RecordingSubscribableDriver Driver(RecordingSubscribableDriverFactory f, string id) => + f.Created.Last(d => d.DriverInstanceId == id); + + /// A fresh replay-opting subscriber for both refs; returns the refs it was replayed. + private string[] ReplayedRefs() + { + 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(); + } + + [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); + + // Positive control before the removal: both values replay. + ReplayedRefs().ShouldBe(new[] { Path2, Path1 }.OrderBy(x => x, StringComparer.Ordinal).ToArray()); + + // 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); + + AwaitAssert( + () => ReplayedRefs().ShouldBe(new[] { Path2 }, "DI-1 was removed: its last Good must not replay; DI-2's must"), + PresenceBudget); + } + + [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"); + + // 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"); + + ReplayedRefs().ShouldBe( + new[] { Path2, Path1 }.OrderBy(x => x, StringComparer.Ordinal).ToArray(), + "a restart is the same driver instance: its last value stays replayable"); + } +}