test(runtime): comms-loss → identical-value recovery contract + silent-resubscribe control (03/S15)

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.

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"
       ]

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;
+
+/// <summary>
+///     <b>The 03/S15 contract test: comms loss → value nodes Bad → the driver's own re-publish clears
+///     them.</b> A real <see cref="DriverInstanceActor"/> is spawned by a real
+///     <see cref="DriverHostActor"/> around a scripted driver, so the whole seam runs: the child's
+///     <c>Reconnecting</c> entry announces comms loss, the host stamps the value nodes
+///     <c>BadNoCommunication</c>, the child reconnects and re-subscribes, and the driver's initial push is
+///     what makes the nodes Good again.
+///
+///     <para>⚠️ <b>Vacuity rule — the pushed value is IDENTICAL to the pre-outage one.</b> 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 <em>different</em> 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.</para>
+///
+///     <para>The falsification control is
+///     <see cref="A_silent_resubscribe_leaves_the_nodes_Bad_and_trips_the_audit"/>: a driver whose
+///     <c>SubscribeAsync</c> 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.</para>
+/// </summary>
+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));
+
+    /// <summary>Audit window measured from the RECONNECT, not from the outage — the initial push lands
+    /// within milliseconds of Connected, so this is generous.</summary>
+    private static readonly TimeSpan AuditDelay = TimeSpan.FromSeconds(2);
+
+    /// <summary>The child's reconnect retry runs on <c>DriverInstanceActor.DefaultReconnectInterval</c>
+    /// (10 s) and the host exposes no seam to shorten it, so every budget here clears that.</summary>
+    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<OpcUaPublishActor.AttributeValueBatchUpdate>(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<OpcUaPublishActor.AttributeStatusBatchUpdate>(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<OpcUaPublishActor.AttributeValueBatchUpdate>(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);
+    }
+
+    /// <summary>
+    ///     <b>The falsification control.</b> Same scenario, but the driver's <c>SubscribeAsync</c> 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.
+    /// </summary>
+    [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<OpcUaPublishActor.AttributeValueBatchUpdate>(PresenceBudget);
+
+        EventFilter.Warning(contains: "still BadNoCommunication").ExpectOne(ReconnectBudget, () =>
+        {
+            host.Tell(new ReconnectDriver("c1", "drv-silent", "test-operator", Guid.NewGuid()));
+            publish.ExpectMsg<OpcUaPublishActor.AttributeStatusBatchUpdate>(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);
+    }
+
+    /// <summary>Hands the host one scripted driver, whatever type/instance it asks for.</summary>
+    private sealed class SingleDriverFactory(IDriver driver) : IDriverFactory
+    {
+        public IDriver? TryCreate(string driverType, string driverInstanceId, string driverConfigJson) => driver;
+
+        public IReadOnlyCollection<string> SupportedTypes { get; } = ["Modbus"];
+    }
+
+    /// <summary>
+    ///     A subscribable driver whose <see cref="SubscribeAsync"/> optionally performs an <b>initial
+    ///     push</b> — the poll engine's <c>forceRaise: true</c> and MQTT's post-STAB-18 behaviour, modelled
+    ///     at the seam the host actually observes. With <see cref="InitialPush"/> null it is the silent
+    ///     shape the control test needs.
+    /// </summary>
+    private sealed class ScriptedSubscribeDriver : IDriver, ISubscribable
+    {
+        private readonly StubHandle _handle = new();
+        private int _subscribeCount;
+
+        public event EventHandler<DataChangeEventArgs>? OnDataChange;
+
+        /// <summary>The value re-published on every (re)subscribe, or null for a silent driver.</summary>
+        public (string Ref, object? Value, uint StatusCode)? InitialPush { get; init; }
+
+        /// <summary>The instance id the child actor adopts — <c>DriverInstanceActor</c> 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.</summary>
+        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<ISubscriptionHandle> SubscribeAsync(
+            IReadOnlyList<string> 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<ISubscriptionHandle>(_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));
+        }
+    }
+
+    /// <summary>Tag-filtered counter recorder — see the note in
+    /// <c>DriverHostActorCommsLossValueNodesTests</c> on why the instrument name alone is not enough.</summary>
+    private sealed class MeterRecorder : IDisposable
+    {
+        private readonly MeterListener _listener;
+        private readonly List<(long Value, KeyValuePair<string, object?>[] 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<long>((_, 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();
+    }
+}

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
     /// <param name="rev">The revision hash the deployment is sealed under.</param>
     /// <param name="rawTags">The raw tags to materialise, one per (driver, device, tag).</param>
     /// <param name="unsRefs">UNS references projecting a raw tag (by <c>BackingTag</c>) into an equipment.</param>
+    /// <param name="driversEnabled">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 <c>DriverInstanceActor</c> sets it and supplies a driver factory.</param>
     /// <returns>The seeded deployment's id.</returns>
     public static DeploymentId SeedV3Deployment(
         IDbContextFactory<OtOpcUaConfigDbContext> 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
     /// <param name="commsRecoveryAuditDelay">Optional 03/S15 audit delay — how long after a reconnect the
     /// host checks for refs still latched <c>BadNoCommunication</c>. Null keeps the production default,
     /// which is far longer than any test's lifetime.</param>
+    /// <param name="driverFactory">Optional factory the host builds its driver children from. Null (the
+    /// default) leaves the null factory, which — with <c>driversEnabled: false</c> — means no child is
+    /// spawned at all.</param>
     /// <returns>The host actor plus the publish and mux probes wired into it.</returns>
     public static (IActorRef Actor, TestProbe Publish, TestProbe Mux) SpawnHostAndApply(
         TestKitBase kit, IDbContextFactory<OtOpcUaConfigDbContext> 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<OpcUaPublishActor.RebuildAddressSpace>(RuntimeActorTestBase.PresenceBudget);
@@ -151,7 +160,8 @@ public static class V3ArtifactSeeder
 
     private static (IActorRef Actor, TestProbe Mux, TestProbe Coordinator) Spawn(
         TestKitBase kit, IDbContextFactory<OtOpcUaConfigDbContext> 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<string> { "driver" },
             dependencyMux: mux.Ref,
             opcUaPublishActor: opcUaPublishActor,