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
This commit is contained in:
Joseph Doherty
+119
@@ -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;
|
||||
|
||||
/// <summary>
|
||||
/// The dependency mux retains a ref's last value for replay-opting subscribers and evicts a driver's values on
|
||||
/// <see cref="DependencyMuxActor.SourceRemoved"/>. 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. <c>DriverHostActor.StopChild</c> 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.
|
||||
/// </summary>
|
||||
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),
|
||||
];
|
||||
|
||||
/// <summary>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.</summary>
|
||||
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<string> { "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<string>();
|
||||
while (seen.Count < 2)
|
||||
foreach (var item in holder.ExpectMsg<VirtualTagActor.DependencyValueBatch>(PresenceBudget).Items)
|
||||
seen.Add(item.TagId);
|
||||
|
||||
_mux = mux;
|
||||
return (host, coordinator, factory);
|
||||
}
|
||||
|
||||
private IActorRef _mux = ActorRefs.Nobody;
|
||||
private Microsoft.EntityFrameworkCore.IDbContextFactory<ZB.MOM.WW.OtOpcUa.Configuration.OtOpcUaConfigDbContext> _db = null!;
|
||||
|
||||
private static RecordingSubscribableDriver Driver(RecordingSubscribableDriverFactory f, string id) =>
|
||||
f.Created.Last(d => d.DriverInstanceId == id);
|
||||
|
||||
/// <summary>A fresh replay-opting subscriber for both refs; returns the refs it was replayed.</summary>
|
||||
private string[] ReplayedRefs()
|
||||
{
|
||||
var probe = CreateTestProbe();
|
||||
_mux.Tell(new DependencyMuxActor.RegisterInterest(new[] { Path1, Path2 }, probe.Ref, ReplayRetained: true));
|
||||
return probe.ExpectMsg<VirtualTagActor.DependencyValueBatch>(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");
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user