From ae97b32c82ca3ff9c51ec9249a69b6c0560fcd97 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 20 Aug 2026 10:27:48 -0400 Subject: [PATCH] =?UTF-8?q?test(runtime):=20comms-loss=20=E2=86=92=20ident?= =?UTF-8?q?ical-value=20recovery=20contract=20+=20silent-resubscribe=20con?= =?UTF-8?q?trol=20(03/S15)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../plans/R3-02-stale-good-family-plan.md | 24 ++ ...R3-02-stale-good-family-plan.md.tasks.json | 2 +- .../DriverHostCommsLossRecoveryTests.cs | 237 ++++++++++++++++++ .../Harness/V3ArtifactSeeder.cs | 21 +- 4 files changed, 278 insertions(+), 6 deletions(-) create mode 100644 tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostCommsLossRecoveryTests.cs diff --git a/archreview/plans/R3-02-stale-good-family-plan.md b/archreview/plans/R3-02-stale-good-family-plan.md index f24af049..0c497741 100644 --- a/archreview/plans/R3-02-stale-good-family-plan.md +++ b/archreview/plans/R3-02-stale-good-family-plan.md @@ -707,3 +707,27 @@ test could only pass once the production change landed. `MeterListener` sees every measurement in the process). - The latch counts **RawPaths**, so the raw node and its UNS mirrors report as one ref — the operator is told how many tags did not come back, not how many nodes. + +### T10 — S15 end-to-end recovery contract + silent-resubscribe control + +- RED: `--filter "…DriverHostCommsLossRecoveryTests"` ⇒ **2 failed**, `Timeout 00:00:15 while waiting for + a message of type …AttributeValueBatchUpdate` (before `AttributeStatusBatchUpdate` even existed in the + first draft, and again while the fake's instance id disagreed — see below). +- GREEN: 2/2 (33 s wall clock — see the reconnect-interval note). `Runtime.Tests` `~Drivers` filter: + 333 passed / 3 skipped. +- **Two harness facts worth keeping** (both cost a red run to find): + - `DriverInstanceActor` takes its instance id from **`driver.DriverInstanceId`**, not from the spec the + host spawned it with. A fake returning a different id publishes under a key the host's ref map does + not contain, and every value is silently dropped as unmapped. The fake now takes the id. + - The fake's push-on-subscribe fires on the FIRST subscribe too, so the pre-outage value must come from + that push rather than from a separate seeding call — otherwise the probe holds an extra batch. +- **Deviation:** the plan sketched driving the outage with the #527 watchdog (`commsLossTimeout` 200 ms). + The host spawns its children with the **default** `reconnectInterval`/`healthPollInterval` and exposes + no seam for either, so the outage is driven by the production `ReconnectDriver` (operator Force + Reconnect) command instead — same `Reconnecting` entry, same `ConnectivityChanged(false)`, and it does + not touch `HandleRestartDriver` (R3-03's). The watchdog's own trip condition stays covered by + `DriverInstanceActorCommsLossWatchdogTests`. The 10 s `DefaultReconnectInterval` is why these two tests + take ~15 s each. +- The identical-value rule is honoured: the driver pushes **5.0 Good** before and after the outage, so the + recovery assertion cannot be satisfied by a value change. The control (`InitialPush = null`) asserts + `ExpectNoMsg` — no Good ever arrives — and that the audit fires with exactly 1 latched ref. diff --git a/archreview/plans/R3-02-stale-good-family-plan.md.tasks.json b/archreview/plans/R3-02-stale-good-family-plan.md.tasks.json index 59c5fb58..ab034166 100644 --- a/archreview/plans/R3-02-stale-good-family-plan.md.tasks.json +++ b/archreview/plans/R3-02-stale-good-family-plan.md.tasks.json @@ -71,7 +71,7 @@ { "id": "T10", "subject": "S15 end-to-end recovery contract test: real DriverInstanceActor + scripted fake with IDENTICAL-value initial push; silent-resubscribe falsification control", - "status": "pending", + "status": "done", "blockedBy": [ "T9" ] diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostCommsLossRecoveryTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostCommsLossRecoveryTests.cs new file mode 100644 index 00000000..81fb3fbb --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostCommsLossRecoveryTests.cs @@ -0,0 +1,237 @@ +using System.Diagnostics.Metrics; +using Akka.Actor; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Commons.Messages.Admin; +using ZB.MOM.WW.OtOpcUa.Commons.Observability; +using ZB.MOM.WW.OtOpcUa.Commons.OpcUa; +using ZB.MOM.WW.OtOpcUa.Commons.Types; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; +using ZB.MOM.WW.OtOpcUa.Runtime.Drivers; +using ZB.MOM.WW.OtOpcUa.Runtime.OpcUa; +using ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness; +using static ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness.V3ArtifactSeeder; + +namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Drivers; + +/// +/// The 03/S15 contract test: comms loss → value nodes Bad → the driver's own re-publish clears +/// them. A real is spawned by a real +/// around a scripted driver, so the whole seam runs: the child's +/// Reconnecting entry announces comms loss, the host stamps the value nodes +/// BadNoCommunication, the child reconnects and re-subscribes, and the driver's initial push is +/// what makes the nodes Good again. +/// +/// ⚠️ Vacuity rule — the pushed value is IDENTICAL to the pre-outage one. The host writes +/// nothing Good on reconnect by design, so recovery depends entirely on a re-publish landing; and every +/// dedup on that path (a driver's last-value cache, a consumer's last-published memory) keys on value. +/// A recovery test whose driver pushes a different value therefore passes against a build with +/// no force-republish anywhere — the value change alone defeats every dedup — and is not this +/// contract's guard. The same-value push is the whole point. +/// +/// The falsification control is +/// : a driver whose +/// SubscribeAsync pushes nothing (the pre-STAB-18 MQTT shape) leaves the nodes Bad and is caught +/// only by the audit. It proves the test above can fail. +/// +public sealed class DriverHostCommsLossRecoveryTests : RuntimeActorTestBase +{ + private static readonly NodeId TestNode = NodeId.Parse("driver-commsloss-recovery"); + private static readonly RevisionHash RevA = RevisionHash.Parse(new string('d', 64)); + + /// Audit window measured from the RECONNECT, not from the outage — the initial push lands + /// within milliseconds of Connected, so this is generous. + private static readonly TimeSpan AuditDelay = TimeSpan.FromSeconds(2); + + /// The child's reconnect retry runs on DriverInstanceActor.DefaultReconnectInterval + /// (10 s) and the host exposes no seam to shorten it, so every budget here clears that. + private static readonly TimeSpan ReconnectBudget = TimeSpan.FromSeconds(25); + + private const string SpeedRef = "Plant/Modbus/dev1/speed"; + private const string UnsSpeedNode = "filling/line1/station1/speed"; + + [Fact] + public void An_identical_value_pushed_on_resubscribe_recovers_the_nodes() + { + var driver = new ScriptedSubscribeDriver { InstanceId = "drv-recover", InitialPush = (SpeedRef, 5.0, OpcUaStatus.Good) }; + var (host, publish) = Arrange(driver, "drv-recover"); + using var meter = new MeterRecorder("otopcua.driver.comms_latched_refs"); + + // Steady state: the driver's FIRST subscribe already pushed 5 Good, and the host fanned it to + // raw + UNS. (The push on subscribe is the behaviour under test, so it is also how the pre-outage + // value gets there — no separate seeding call, which would leave an extra batch in the probe.) + var first = publish.ExpectMsg(PresenceBudget); + first.Items.Select(i => i.NodeId).ShouldBe(new[] { SpeedRef, UnsSpeedNode }, ignoreOrder: true); + first.Items.ShouldAllBe(i => Equals(i.Value, 5.0)); + + // The outage: the child re-enters Reconnecting and announces comms loss to the host. + host.Tell(new ReconnectDriver("c1", "drv-recover", "test-operator", Guid.NewGuid())); + + var staled = publish.ExpectMsg(ReconnectBudget); + staled.Items.Select(i => i.NodeId).ShouldBe(new[] { SpeedRef, UnsSpeedNode }, ignoreOrder: true); + staled.Items.ShouldAllBe(i => i.StatusCode == OpcUaStatus.BadNoCommunication); + + // Recovery: the reconnect re-subscribes, and the driver pushes the SAME value it held before. + var recovered = publish.ExpectMsg(ReconnectBudget); + recovered.Items.Select(i => i.NodeId).ShouldBe(new[] { SpeedRef, UnsSpeedNode }, ignoreOrder: true); + recovered.Items.ShouldAllBe(i => i.StatusCode == OpcUaStatus.Good); + recovered.Items.ShouldAllBe(i => Equals(i.Value, 5.0)); + + // …and the audit finds nothing outstanding. + EventFilter.Warning(contains: "still BadNoCommunication") + .Expect(0, AuditDelay + TimeSpan.FromSeconds(3), () => { }); + meter.SumWithTag("driver", "drv-recover").ShouldBe(0); + } + + /// + /// The falsification control. Same scenario, but the driver's SubscribeAsync pushes + /// nothing — exactly what MQTT did before 05/STAB-18, and what any push driver without an initial + /// push does. The nodes stay Bad (the host writes no Good of its own), and the ONLY thing that says + /// so is the recovery audit. + /// + [Fact] + public void A_silent_resubscribe_leaves_the_nodes_Bad_and_trips_the_audit() + { + var driver = new ScriptedSubscribeDriver { InstanceId = "drv-silent", InitialPush = null }; + var (host, publish) = Arrange(driver, "drv-silent"); + using var meter = new MeterRecorder("otopcua.driver.comms_latched_refs"); + + driver.FireDataChange(SpeedRef, 5.0, OpcUaStatus.Good); + publish.ExpectMsg(PresenceBudget); + + EventFilter.Warning(contains: "still BadNoCommunication").ExpectOne(ReconnectBudget, () => + { + host.Tell(new ReconnectDriver("c1", "drv-silent", "test-operator", Guid.NewGuid())); + publish.ExpectMsg(ReconnectBudget); + + // Nothing Good ever arrives — assert it rather than merely not looking. + publish.ExpectNoMsg(TimeSpan.FromSeconds(1)); + }); + + // One ref (the raw tag); its UNS mirror shares it. + AwaitCondition(() => meter.SumWithTag("driver", "drv-silent") == 1, TimeSpan.FromSeconds(5)); + driver.SubscribeCount.ShouldBeGreaterThanOrEqualTo(2, + "the child must actually have re-subscribed — otherwise this proves nothing about the push"); + } + + private (IActorRef Host, Akka.TestKit.TestProbe Publish) Arrange(ScriptedSubscribeDriver driver, string driverId) + { + var db = NewInMemoryDbFactory(); + var deploymentId = SeedV3Deployment(db, RevA, + rawTags: new[] + { + (Driver: driverId, DriverName: "Modbus", Device: "dev1", Tag: "speed", DataType: "Double", Writable: false), + }, + unsRefs: new[] { (Area: "filling", Line: "line1", Equip: "station1", BackingTag: "speed", Effective: (string?)null) }, + driversEnabled: true); + + var (host, publish, _) = SpawnHostAndApply( + this, db, TestNode, deploymentId, RevA, + commsRecoveryAuditDelay: AuditDelay, + driverFactory: new SingleDriverFactory(driver)); + + // The child is spawned + initialised by the apply; wait for its first subscribe so the scripted + // pushes below are attributed to a live subscription. + AwaitCondition(() => driver.SubscribeCount >= 1, PresenceBudget); + return (host, publish); + } + + /// Hands the host one scripted driver, whatever type/instance it asks for. + private sealed class SingleDriverFactory(IDriver driver) : IDriverFactory + { + public IDriver? TryCreate(string driverType, string driverInstanceId, string driverConfigJson) => driver; + + public IReadOnlyCollection SupportedTypes { get; } = ["Modbus"]; + } + + /// + /// A subscribable driver whose optionally performs an initial + /// push — the poll engine's forceRaise: true and MQTT's post-STAB-18 behaviour, modelled + /// at the seam the host actually observes. With null it is the silent + /// shape the control test needs. + /// + private sealed class ScriptedSubscribeDriver : IDriver, ISubscribable + { + private readonly StubHandle _handle = new(); + private int _subscribeCount; + + public event EventHandler? OnDataChange; + + /// The value re-published on every (re)subscribe, or null for a silent driver. + public (string Ref, object? Value, uint StatusCode)? InitialPush { get; init; } + + /// The instance id the child actor adopts — DriverInstanceActor takes it from the + /// DRIVER, not from the spec, and the host's ref map is keyed by the spec's id. They must agree or + /// every published value is dropped as unmapped. + public required string InstanceId { get; init; } + + public int SubscribeCount => Volatile.Read(ref _subscribeCount); + + public string DriverInstanceId => InstanceId; + + public string DriverType => "Modbus"; + + 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; + + // Stable health: the #527 watchdog must not fire on its own here — the outage in these tests is + // driven explicitly, and a health-driven trip would make the timing non-deterministic. + public DriverHealth GetHealth() => new(DriverState.Healthy, DateTime.UtcNow, null); + + public Task SubscribeAsync( + IReadOnlyList fullReferences, TimeSpan publishingInterval, CancellationToken cancellationToken) + { + Interlocked.Increment(ref _subscribeCount); + if (InitialPush is { } push && fullReferences.Contains(push.Ref)) + FireDataChange(push.Ref, push.Value, push.StatusCode); + + return Task.FromResult(_handle); + } + + public Task UnsubscribeAsync(ISubscriptionHandle handle, CancellationToken cancellationToken) => Task.CompletedTask; + + public void FireDataChange(string fullRef, object? value, uint statusCode) + { + var snapshot = new DataValueSnapshot(value, statusCode, DateTime.UtcNow, DateTime.UtcNow); + OnDataChange?.Invoke(this, new DataChangeEventArgs(_handle, fullRef, snapshot)); + } + } + + /// Tag-filtered counter recorder — see the note in + /// DriverHostActorCommsLossValueNodesTests on why the instrument name alone is not enough. + private sealed class MeterRecorder : IDisposable + { + private readonly MeterListener _listener; + private readonly List<(long Value, KeyValuePair[] Tags)> _measurements = new(); + private readonly object _gate = new(); + + public MeterRecorder(string instrumentName) + { + _listener = new MeterListener + { + InstrumentPublished = (instrument, listener) => + { + if (instrument.Meter.Name == OtOpcUaTelemetry.MeterName && instrument.Name == instrumentName) + listener.EnableMeasurementEvents(instrument); + }, + }; + _listener.SetMeasurementEventCallback((_, value, tags, _) => + { + lock (_gate) _measurements.Add((value, tags.ToArray())); + }); + _listener.Start(); + } + + public long SumWithTag(string key, string value) + { + lock (_gate) + return _measurements.Where(m => m.Tags.Any(t => t.Key == key && Equals(t.Value, value))) + .Sum(m => m.Value); + } + + public void Dispose() => _listener.Dispose(); + } +} diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Harness/V3ArtifactSeeder.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Harness/V3ArtifactSeeder.cs index 0884a0f6..aa620963 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Harness/V3ArtifactSeeder.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Harness/V3ArtifactSeeder.cs @@ -9,6 +9,7 @@ using ZB.MOM.WW.OtOpcUa.Commons.Types; using ZB.MOM.WW.OtOpcUa.Configuration; using ZB.MOM.WW.OtOpcUa.Configuration.Entities; using ZB.MOM.WW.OtOpcUa.Configuration.Enums; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; using ZB.MOM.WW.OtOpcUa.Runtime.Drivers; using ZB.MOM.WW.OtOpcUa.Runtime.OpcUa; @@ -36,16 +37,20 @@ public static class V3ArtifactSeeder /// The revision hash the deployment is sealed under. /// The raw tags to materialise, one per (driver, device, tag). /// UNS references projecting a raw tag (by BackingTag) into an equipment. + /// Whether the seeded DriverInstance rows are Enabled — i.e. whether the + /// host SPAWNS a child around them. False (the default) keeps the routing suites child-free; a suite + /// that needs a real DriverInstanceActor sets it and supplies a driver factory. /// The seeded deployment's id. public static DeploymentId SeedV3Deployment( IDbContextFactory db, RevisionHash rev, (string Driver, string DriverName, string Device, string Tag, string DataType, bool Writable)[] rawTags, - (string Area, string Line, string Equip, string BackingTag, string? Effective)[] unsRefs) + (string Area, string Line, string Equip, string BackingTag, string? Effective)[] unsRefs, + bool driversEnabled = false) { var drivers = rawTags .Select(t => (t.Driver, t.DriverName)) .Distinct() - .Select(d => new { DriverInstanceId = d.Driver, RawFolderId = "rf-plant", Name = d.DriverName, DriverType = "Modbus", DriverConfig = "{}", ClusterId = "c1", Enabled = false }) + .Select(d => new { DriverInstanceId = d.Driver, RawFolderId = "rf-plant", Name = d.DriverName, DriverType = "Modbus", DriverConfig = "{}", ClusterId = "c1", Enabled = driversEnabled }) .ToArray(); var devices = rawTags .Select(t => (t.Driver, t.Device)) @@ -117,14 +122,18 @@ public static class V3ArtifactSeeder /// Optional 03/S15 audit delay — how long after a reconnect the /// host checks for refs still latched BadNoCommunication. Null keeps the production default, /// which is far longer than any test's lifetime. + /// Optional factory the host builds its driver children from. Null (the + /// default) leaves the null factory, which — with driversEnabled: false — means no child is + /// spawned at all. /// The host actor plus the publish and mux probes wired into it. public static (IActorRef Actor, TestProbe Publish, TestProbe Mux) SpawnHostAndApply( TestKitBase kit, IDbContextFactory db, NodeId localNode, DeploymentId deploymentId, RevisionHash rev, - TimeSpan? commsRecoveryAuditDelay = null) + TimeSpan? commsRecoveryAuditDelay = null, + IDriverFactory? driverFactory = null) { var publish = kit.CreateTestProbe(); - var (actor, mux, coordinator) = Spawn(kit, db, localNode, publish.Ref, commsRecoveryAuditDelay); + var (actor, mux, coordinator) = Spawn(kit, db, localNode, publish.Ref, commsRecoveryAuditDelay, driverFactory); Dispatch(actor, coordinator, deploymentId, rev); publish.ExpectMsg(RuntimeActorTestBase.PresenceBudget); @@ -151,7 +160,8 @@ public static class V3ArtifactSeeder private static (IActorRef Actor, TestProbe Mux, TestProbe Coordinator) Spawn( TestKitBase kit, IDbContextFactory db, NodeId localNode, - IActorRef? opcUaPublishActor, TimeSpan? commsRecoveryAuditDelay = null) + IActorRef? opcUaPublishActor, TimeSpan? commsRecoveryAuditDelay = null, + IDriverFactory? driverFactory = null) { var coordinator = kit.CreateTestProbe(); var mux = kit.CreateTestProbe(); @@ -159,6 +169,7 @@ public static class V3ArtifactSeeder var actor = kit.Sys.ActorOf(DriverHostActor.Props( db, localNode, coordinator.Ref, + driverFactory: driverFactory, localRoles: new HashSet { "driver" }, dependencyMux: mux.Ref, opcUaPublishActor: opcUaPublishActor,