diff --git a/archreview/plans/R3-02-stale-good-family-plan.md b/archreview/plans/R3-02-stale-good-family-plan.md index 851e9931..f24af049 100644 --- a/archreview/plans/R3-02-stale-good-family-plan.md +++ b/archreview/plans/R3-02-stale-good-family-plan.md @@ -690,3 +690,20 @@ test could only pass once the production change landed. this method, but that helper is T9's — calling it here would not compile. The call is added in T9, where the latch and its audit timer land together. R3-03 boundary respected: this is a new method plus one branch in `OnDriverConnectivityChanged`; `HandleRestartDriver` and the watchdog are untouched. + +### T9 — recovery audit (`_commsLatched` + timer + `otopcua.driver.comms_latched_refs`) + +- RED: `--filter "…DriverHostActorCommsLossValueNodesTests"` ⇒ **build failure**, `CS1739: The best + overload for 'Props' does not have a parameter named 'commsRecoveryAuditDelay'`. +- GREEN: 6/6 in the suite; 20 passed across it + `DriverHostActorCommsLossTests` + + `DriverHostActorBatchFanOutTests` + `DriverHostActorPrimaryGateTests`. +- **Falsification of the control** (the plan's discipline, applied to the audit rather than to a dedup + key): deleting the prune loop in `ForwardBatchToMux` ⇒ + `A_driver_that_republishes_every_ref_trips_no_audit` **FAILED** with `Received 1 message too many. + Expected 0 messages but received 1 that matched filter [Warning when Message contains "still + BadNoCommunication"]`. Restored, green. So the "no audit fires" assertion is falsifiable and the audit + is not merely dormant. +- Meter assertions filter on a per-test **driver tag**, not on the instrument name alone (#503: a + `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. 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 4ec788d7..59c5fb58 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 @@ -63,7 +63,7 @@ { "id": "T9", "subject": "Recovery audit: _commsLatched per driver, pruned by routed values, CommsRecoveryAudit timer + otopcua.driver.comms_latched_refs meter (dedup-trap tripwire)", - "status": "pending", + "status": "done", "blockedBy": [ "T8" ] diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Observability/OtOpcUaTelemetry.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Observability/OtOpcUaTelemetry.cs index 09950d24..07579b26 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Observability/OtOpcUaTelemetry.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Observability/OtOpcUaTelemetry.cs @@ -106,6 +106,15 @@ public static class OtOpcUaTelemetry /// Inbound operations denied by the Primary data-plane gate (archreview 03/S4) — /// site=write|ack|alarm-emit, reason=secondary|detached|role-unknown. + /// Archreview 03/S15 — tag refs a driver left BadNoCommunication after it reconnected: + /// the host marked them Bad on the outage, and the driver never re-published them. Non-zero means a + /// driver that reconnects without a full re-publish (a push driver missing its initial push, or a + /// value-only dedup swallowing an unchanged value), which leaves those nodes Bad indefinitely. + /// Tagged driver + driver_type. + public static readonly Counter DriverCommsLatchedRefs = + Meter.CreateCounter("otopcua.driver.comms_latched_refs", unit: "{ref}", + description: "Refs still BadNoCommunication after their driver reconnected (driver, driver_type)."); + public static readonly Counter PrimaryGateDenied = Meter.CreateCounter("otopcua.redundancy.primary_gate_denied", unit: "{denial}", description: "Operations denied by the Primary gate (site=write|ack|alarm-emit, reason=secondary|detached|role-unknown)."); 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 56d918d0..2c643dea 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs @@ -82,6 +82,14 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS /// public static readonly TimeSpan DefaultApplyWedgeTimeout = TimeSpan.FromMinutes(5); + /// + /// How long after a driver reconnects the 03/S15 recovery audit waits before reporting refs still + /// latched . Sized so an ordinary poll driver has had + /// several poll cycles — and a slow one at least one — to re-publish before anything is said; the + /// audit's job is to name a driver that will NEVER re-publish, not to race a slow first poll. + /// + public static readonly TimeSpan DefaultCommsRecoveryAuditDelay = TimeSpan.FromSeconds(30); + /// /// Cache key used when an artifact carries no cluster scoping (single-cluster or unscoped /// deployments). A literal rather than an empty string so the row is visibly deliberate when @@ -401,6 +409,27 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS /// ). private readonly TimeSpan _applyWedgeTimeout; + /// How long after a driver reconnects the 03/S15 recovery audit waits before reporting refs + /// still latched (see ). + private readonly TimeSpan _commsRecoveryAuditDelay; + + /// + /// Archreview 03/S15 — RawPaths this host has stamped BadNoCommunication on the value + /// nodes of, per driver, and has not yet seen a value for. Set by + /// , pruned by as each ref + /// re-publishes, and read once by after the driver reconnects. + /// It is a tripwire, not a mechanism. Nothing depends on it being correct for the data + /// plane to behave: the host publishes nothing Good on reconnect either way. What it buys is that a + /// driver which reconnects and then fails to re-publish — the pre-STAB-18 MQTT shape, and what any + /// future push driver without an initial push will do — produces a Warning and a metered count + /// instead of a subscription that silently reports Bad forever. + /// Keyed by RawPath, so a raw node and its UNS mirrors count once: the operator is being told + /// how many tags did not come back, not how many address-space nodes they project onto. The + /// entry is removed as soon as it empties, so the steady-state cost in + /// is one dictionary miss per burst. + /// + private readonly Dictionary> _commsLatched = new(StringComparer.Ordinal); + /// /// Host of this node's LocalDb replication partner — the only node that holds a copy of this /// node's alarm queue, and therefore the only node it may ever stand down in favour of. @@ -449,6 +478,12 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS /// The authenticated principal performing the acknowledge. public sealed record RouteNativeAlarmAck(string ConditionNodeId, string? Comment, string OperatorUser); + /// Self-sent by the 03/S15 audit timer, after a driver + /// reconnected: report any of its refs still latched BadNoCommunication. See + /// . + /// The driver whose recovery is being audited. + public sealed record CommsRecoveryAudit(string DriverInstanceId); + private sealed record ChildEntry(IActorRef Actor, DriverInstanceSpec Spec, bool Stubbed) { // Convenience accessors for sites that don't need the full spec. @@ -514,6 +549,9 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS /// VirtualTag + ScriptedAlarm hosts. Defaults to null (admin-only nodes / tests skip the emit). /// How long the host may sit in before the /// wedge watchdog logs a Warning. Defaults to . + /// Archreview 03/S15: how long after a driver reconnects the + /// recovery audit waits before reporting refs still latched BadNoCommunication. Defaults to + /// ; tests shorten it. /// The Akka.NET used to spawn this actor. public static Props Props( IDbContextFactory? dbFactory, @@ -542,7 +580,8 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS IDriverPrimaryRoleSource? primaryRoleSource = null, IHistorizationRoleView? historizationRoleView = null, Func? sharesHistorizationScopeWithLocal = null, - TimeSpan? applyWedgeTimeout = null) => + TimeSpan? applyWedgeTimeout = null, + TimeSpan? commsRecoveryAuditDelay = null) => // WARNING: this forwarding list is POSITIONAL, and Props.Create compiles it into an // expression tree. Six IActorRef? parameters and several interface-typed ones mean a // mis-ordered argument is usually type-compatible and therefore compiles clean, then binds @@ -554,7 +593,7 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS loggerFactory, scriptRootLogger, scriptedAlarmHostOverride, invokerFactory, driverMemberCountProvider, deploymentArtifactCache, redundancyRoleView, replicationPeerHost, fetchAndCacheMode, artifactFetcher, alarmStateStore, telemetryHub, primaryRoleSource, historizationRoleView, - sharesHistorizationScopeWithLocal, applyWedgeTimeout)); + sharesHistorizationScopeWithLocal, applyWedgeTimeout, commsRecoveryAuditDelay)); /// Initializes a new DriverHostActor with the specified dependencies. /// Database context factory for configuration database access. @@ -613,6 +652,9 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS /// How long the host may sit in before the /// wedge watchdog logs a Warning. Defaults to ; a test shortens /// it so the watchdog is observable in a test's lifetime. + /// Archreview 03/S15: how long after a driver reconnects the + /// recovery audit waits before reporting refs still latched BadNoCommunication. Defaults to + /// ; a test shortens it for the same reason. public DriverHostActor( IDbContextFactory? dbFactory, CommonsNodeId localNode, @@ -640,9 +682,11 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS IDriverPrimaryRoleSource? primaryRoleSource = null, IHistorizationRoleView? historizationRoleView = null, Func? sharesHistorizationScopeWithLocal = null, - TimeSpan? applyWedgeTimeout = null) + TimeSpan? applyWedgeTimeout = null, + TimeSpan? commsRecoveryAuditDelay = null) { _applyWedgeTimeout = applyWedgeTimeout ?? DefaultApplyWedgeTimeout; + _commsRecoveryAuditDelay = commsRecoveryAuditDelay ?? DefaultCommsRecoveryAuditDelay; _historizationRoleView = historizationRoleView; _sharesHistorizationScopeWithLocal = sharesHistorizationScopeWithLocal; _telemetryHub = telemetryHub; @@ -1073,6 +1117,7 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS Receive(ForwardBatchToMux); Receive(ForwardNativeAlarm); Receive(OnDriverConnectivityChanged); + Receive(HandleCommsRecoveryAudit); Receive(HandleDeltaApplied); Receive(HandleApplyResult); Receive(HandleRestartDriver); @@ -1128,6 +1173,7 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS Receive(ForwardBatchToMux); Receive(ForwardNativeAlarm); Receive(OnDriverConnectivityChanged); + Receive(HandleCommsRecoveryAudit); Receive(HandleDeltaApplied); Receive(HandleApplyResult); Receive(HandleRestartDriver); @@ -1168,6 +1214,15 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS // One envelope per burst; the mux filters per item (it registered a batch receive in Task 9). _dependencyMux?.Tell(msg); + // 03/S15 recovery audit: clear whatever this burst re-published from the driver's Bad latch. Guarded + // by ONE dictionary probe per burst — with no outage outstanding (the steady state) this is a single + // miss, not per-item work. + if (_commsLatched.ContainsKey(msg.DriverInstanceId)) + { + for (var i = 0; i < msg.Items.Count; i++) + TrackCommsLatchPrune(msg.DriverInstanceId, msg.Items[i].FullReference); + } + if (_opcUaPublishActor is null) return; // Indexed for-loop, not foreach: IReadOnlyList enumerates through an interface @@ -1266,6 +1321,12 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS // 03/S15: and the EXTERNAL data plane, which #481 stopped one seam short of. FanCommsLossToValueNodes(msg.DriverInstanceId); } + else + { + // 03/S15: nothing Good is published here — instead, check LATER whether the driver republished + // what this host marked Bad. See HandleCommsRecoveryAudit. + TrackCommsLatchArmAudit(msg.DriverInstanceId); + } if (_opcUaPublishActor is null) return; @@ -1352,12 +1413,105 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS if (writes.Count == 0) return; _opcUaPublishActor.Tell(new OpcUa.OpcUaPublishActor.AttributeStatusBatchUpdate(writes)); + TrackCommsLatchSet(driverInstanceId, entries); _log.Info( "DriverHost {Node}: driver {Driver} disconnected — {Count} value node(s) marked BadNoCommunication (value retained)", _localNode, driverInstanceId, writes.Count); } + /// + /// Records every RawPath just latched BadNoCommunication for + /// (03/S15's recovery audit — see ). Replaces any previous latch wholesale: + /// a second outage re-arms from the current ref set rather than accumulating refs a redeploy removed. + /// + /// The driver that just went comms-lost. + /// Its value refs, from the per-driver index. + private void TrackCommsLatchSet( + string driverInstanceId, + List<(string RawPath, HashSet Nodes)> entries) + { + var latched = new HashSet(entries.Count, StringComparer.Ordinal); + foreach (var (rawPath, _) in entries) latched.Add(rawPath); + _commsLatched[driverInstanceId] = latched; + + // A previous reconnect's audit is now meaningless — this driver is down again. + Timers.Cancel(CommsAuditTimerKey(driverInstanceId)); + } + + /// + /// Clears from 's latch, and drops + /// the driver's entry once it empties. Called per routed value item, so the steady-state cost with no + /// latch outstanding is one dictionary miss per burst — not per item. + /// + /// The publishing driver. + /// The RawPath that just published a value. + private void TrackCommsLatchPrune(string driverInstanceId, string fullReference) + { + if (!_commsLatched.TryGetValue(driverInstanceId, out var latched)) return; + if (!latched.Remove(fullReference)) return; + if (latched.Count == 0) _commsLatched.Remove(driverInstanceId); + } + + /// + /// Arms the recovery audit for a driver that has just reconnected: after + /// , reports whatever is + /// still latched. Nothing is armed when the driver has no latch — it never went Bad here. + /// + /// The driver that just reconnected. + private void TrackCommsLatchArmAudit(string driverInstanceId) + { + if (!_commsLatched.ContainsKey(driverInstanceId)) return; + Timers.StartSingleTimer( + CommsAuditTimerKey(driverInstanceId), + new CommsRecoveryAudit(driverInstanceId), + _commsRecoveryAuditDelay); + } + + private static string CommsAuditTimerKey(string driverInstanceId) => $"comms-audit:{driverInstanceId}"; + + /// + /// The 03/S15 tripwire firing: refs this host latched BadNoCommunication that the driver has + /// still not re-published after it reconnected. + /// + /// + /// + /// It reports, it does not repair. Writing Good here would assert a quality nothing has + /// observed — the very thing this whole path refuses to do. What the operator gets instead is a + /// named driver and a count, plus the otopcua.driver.comms_latched_refs counter, which is + /// the operator-visible signature of a driver that reconnects without re-publishing (the + /// pre-STAB-18 MQTT shape, or any future push driver that forgets its initial push). + /// + /// + /// The latch is cleared afterwards either way: it has been reported, and leaving it would make + /// the next outage's audit report refs from this one. + /// + /// + /// Names the driver whose audit is due. + private void HandleCommsRecoveryAudit(CommsRecoveryAudit msg) + { + if (!_commsLatched.TryGetValue(msg.DriverInstanceId, out var latched) || latched.Count == 0) + { + _commsLatched.Remove(msg.DriverInstanceId); + return; + } + + var driverType = _children.TryGetValue(msg.DriverInstanceId, out var child) ? child.DriverType : "unknown"; + OtOpcUaTelemetry.DriverCommsLatchedRefs.Add( + latched.Count, + new KeyValuePair("driver", msg.DriverInstanceId), + new KeyValuePair("driver_type", driverType)); + + _log.Warning( + "DriverHost {Node}: driver {Driver} ({Type}) — {Count} ref(s) still BadNoCommunication {Delay} after reconnect; " + + "the driver did not re-publish them (missing initial push, or a value-only dedup swallowed an unchanged value). " + + "First: {First}", + _localNode, msg.DriverInstanceId, driverType, latched.Count, _commsRecoveryAuditDelay, + latched.OrderBy(r => r, StringComparer.Ordinal).First()); + + _commsLatched.Remove(msg.DriverInstanceId); + } + /// /// #481 (Layer 4) — push a synthetic into the dependency /// mux for every value ref the just-disconnected driver owns, so scripted alarms and virtual tags see @@ -1807,6 +1961,9 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS // A child connectivity transition while the host is Stale has no live address space to annotate — drop // it (the post-recovery rebuild re-materialises conditions, and the child re-announces on its next entry). Receive(_ => { }); + // Likewise the 03/S15 recovery audit: a timer armed before the host went Stale has nothing to report + // on (no latch was set while Stale, and the rebuild re-establishes every node's status anyway). + Receive(_ => { }); // A late DeltaApplied (an apply completed just before the DB went Stale) — re-register the driver's // mux adapter anyway; it simply re-reads the driver's current refs (harmless, no DB access). Receive(HandleDeltaApplied); diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorCommsLossValueNodesTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorCommsLossValueNodesTests.cs index a6e2b528..70f6cd43 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorCommsLossValueNodesTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorCommsLossValueNodesTests.cs @@ -1,6 +1,8 @@ +using System.Diagnostics.Metrics; using Akka.Actor; using Shouldly; using Xunit; +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.Runtime.Drivers; @@ -32,6 +34,17 @@ public sealed class DriverHostActorCommsLossValueNodesTests : RuntimeActorTestBa { private static readonly NodeId TestNode = NodeId.Parse("driver-commsloss-values"); private static readonly RevisionHash RevA = RevisionHash.Parse(new string('c', 64)); + private static readonly DateTime TsA = new(2026, 8, 20, 9, 0, 0, DateTimeKind.Utc); + + /// Audit delay used by the T9 cases — long enough that a same-turn prune wins the race, + /// short enough to observe. + private static readonly TimeSpan AuditDelay = TimeSpan.FromMilliseconds(400); + + // Distinct driver ids per metered test: the meter tag is what the recorder filters on, and a + // MeterListener sees every measurement in the process — including a parallel test's (#503). + private const string SilentDriver = "drv-audit-silent"; + private const string RepublishingDriver = "drv-audit-republish"; + private const string PartialDriver = "drv-audit-partial"; /// /// A disconnect produces ONE status batch naming exactly the disconnected driver's raw nodes plus @@ -102,6 +115,122 @@ public sealed class DriverHostActorCommsLossValueNodesTests : RuntimeActorTestBa publish.ExpectNoMsg(TimeSpan.FromMilliseconds(500)); } + // --------------------------------------------------------------------------------- + // T9 — the recovery audit (the dedup-trap tripwire) + // --------------------------------------------------------------------------------- + + /// + /// The host publishes nothing Good on reconnect, so the only way a stuck-Bad node becomes visible + /// is this audit: refs still latched some time after the driver reconnected are logged and metered. + /// Here the driver re-publishes nothing at all — the pre-STAB-18 MQTT shape, and the exact failure + /// any future push driver without an initial push would produce. + /// + [Fact] + public void Refs_still_latched_after_a_reconnect_are_logged_and_metered() + { + var db = NewInMemoryDbFactory(); + var deploymentId = SeedV3Deployment(db, RevA, + rawTags: new[] + { + (Driver: SilentDriver, DriverName: "Modbus", Device: "dev1", Tag: "speed", DataType: "Double", Writable: false), + (Driver: SilentDriver, DriverName: "Modbus", Device: "dev1", Tag: "temp", DataType: "Double", Writable: false), + }, + unsRefs: new[] { (Area: "filling", Line: "line1", Equip: "station1", BackingTag: "speed", Effective: (string?)null) }); + + var (actor, publish, _) = SpawnHostAndApply(this, db, TestNode, deploymentId, RevA, AuditDelay); + using var meter = new MeterRecorder("otopcua.driver.comms_latched_refs"); + + actor.Tell(new DriverInstanceActor.ConnectivityChanged(SilentDriver, Connected: false)); + publish.ExpectMsg(PresenceBudget); + + EventFilter.Warning(contains: "still BadNoCommunication").ExpectOne(TimeSpan.FromSeconds(10), () => + actor.Tell(new DriverInstanceActor.ConnectivityChanged(SilentDriver, Connected: true))); + + // Two REFS, not three nodes: the latch is keyed by RawPath, and the UNS mirror shares its + // backing tag's ref — the operator is being told how many tags did not come back, not how many + // address-space nodes they project onto. + AwaitCondition(() => meter.WithTag("driver", SilentDriver) > 0, TimeSpan.FromSeconds(5)); + meter.SumWithTag("driver", SilentDriver).ShouldBe(2); + } + + /// + /// The control. A driver that DOES re-publish every latched ref on reconnect trips nothing — + /// without this, the test above would pass against an audit that fires unconditionally, which would + /// make the Warning noise rather than signal. + /// + [Fact] + public void A_driver_that_republishes_every_ref_trips_no_audit() + { + var db = NewInMemoryDbFactory(); + var deploymentId = SeedV3Deployment(db, RevA, + rawTags: new[] + { + (Driver: RepublishingDriver, DriverName: "Modbus", Device: "dev1", Tag: "speed", DataType: "Double", Writable: false), + (Driver: RepublishingDriver, DriverName: "Modbus", Device: "dev1", Tag: "temp", DataType: "Double", Writable: false), + }, + unsRefs: new[] { (Area: "filling", Line: "line1", Equip: "station1", BackingTag: "speed", Effective: (string?)null) }); + + var (actor, publish, _) = SpawnHostAndApply(this, db, TestNode, deploymentId, RevA, AuditDelay); + using var meter = new MeterRecorder("otopcua.driver.comms_latched_refs"); + + actor.Tell(new DriverInstanceActor.ConnectivityChanged(RepublishingDriver, Connected: false)); + publish.ExpectMsg(PresenceBudget); + + EventFilter.Warning(contains: "still BadNoCommunication").Expect(0, TimeSpan.FromSeconds(5), () => + { + actor.Tell(new DriverInstanceActor.ConnectivityChanged(RepublishingDriver, Connected: true)); + + // The driver's own re-publish — identical values, which is precisely the case a value-only + // dedup would swallow; here it is the driver that force-raised, so both refs are pruned. + actor.Tell(new DriverInstanceActor.AttributeValueBatchPublished(RepublishingDriver, new[] + { + new DriverInstanceActor.AttributeValueItem("Plant/Modbus/dev1/speed", 42.0, OpcUaStatus.Good, TsA), + new DriverInstanceActor.AttributeValueItem("Plant/Modbus/dev1/temp", 7.5, OpcUaStatus.Good, TsA), + })); + publish.ExpectMsg(PresenceBudget); + }); + + meter.WithTag("driver", RepublishingDriver).ShouldBe(0); + } + + /// + /// A partial recovery is still reported, and reports only what is missing: one ref comes back, the + /// other does not, and the audit counts exactly one. An audit that reported the whole latch (or + /// nothing at all once anything arrived) would be useless on precisely the mixed case an operator + /// needs it for. + /// + [Fact] + public void A_partial_recovery_reports_only_the_refs_that_did_not_come_back() + { + var db = NewInMemoryDbFactory(); + var deploymentId = SeedV3Deployment(db, RevA, + rawTags: new[] + { + (Driver: PartialDriver, DriverName: "Modbus", Device: "dev1", Tag: "speed", DataType: "Double", Writable: false), + (Driver: PartialDriver, DriverName: "Modbus", Device: "dev1", Tag: "temp", DataType: "Double", Writable: false), + }, + unsRefs: Array.Empty<(string, string, string, string, string?)>()); + + var (actor, publish, _) = SpawnHostAndApply(this, db, TestNode, deploymentId, RevA, AuditDelay); + using var meter = new MeterRecorder("otopcua.driver.comms_latched_refs"); + + actor.Tell(new DriverInstanceActor.ConnectivityChanged(PartialDriver, Connected: false)); + publish.ExpectMsg(PresenceBudget); + + EventFilter.Warning(contains: "still BadNoCommunication").ExpectOne(TimeSpan.FromSeconds(10), () => + { + actor.Tell(new DriverInstanceActor.ConnectivityChanged(PartialDriver, Connected: true)); + actor.Tell(new DriverInstanceActor.AttributeValueBatchPublished(PartialDriver, new[] + { + new DriverInstanceActor.AttributeValueItem("Plant/Modbus/dev1/speed", 42.0, OpcUaStatus.Good, TsA), + })); + publish.ExpectMsg(PresenceBudget); + }); + + AwaitCondition(() => meter.WithTag("driver", PartialDriver) > 0, TimeSpan.FromSeconds(5)); + meter.SumWithTag("driver", PartialDriver).ShouldBe(1, "only the ref that never came back is reported"); + } + /// /// A driver that owns no value refs produces no message at all — an empty batch would cost a hop /// and a sink call for nothing on every flap of an alarm-only or freshly-added driver. @@ -123,4 +252,47 @@ public sealed class DriverHostActorCommsLossValueNodesTests : RuntimeActorTestBa publish.ExpectNoMsg(TimeSpan.FromMilliseconds(500)); } + + /// + /// Sums one counter's measurements, filtered by tag. A observes + /// every measurement in the process, so filtering on instrument name alone would fold in a parallel + /// test's (#503) — each test here therefore uses its own driver id and asserts on that tag. + /// + 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 int WithTag(string key, string value) + { + lock (_gate) return _measurements.Count(m => m.Tags.Any(t => t.Key == key && Equals(t.Value, value))); + } + + 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 4ffc9a1d..0884a0f6 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 @@ -114,13 +114,17 @@ public static class V3ArtifactSeeder /// The host's local node id. /// The deployment to dispatch. /// The revision hash the deployment was sealed under. + /// 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. /// 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) + NodeId localNode, DeploymentId deploymentId, RevisionHash rev, + TimeSpan? commsRecoveryAuditDelay = null) { var publish = kit.CreateTestProbe(); - var (actor, mux, coordinator) = Spawn(kit, db, localNode, publish.Ref); + var (actor, mux, coordinator) = Spawn(kit, db, localNode, publish.Ref, commsRecoveryAuditDelay); Dispatch(actor, coordinator, deploymentId, rev); publish.ExpectMsg(RuntimeActorTestBase.PresenceBudget); @@ -147,7 +151,7 @@ public static class V3ArtifactSeeder private static (IActorRef Actor, TestProbe Mux, TestProbe Coordinator) Spawn( TestKitBase kit, IDbContextFactory db, NodeId localNode, - IActorRef? opcUaPublishActor) + IActorRef? opcUaPublishActor, TimeSpan? commsRecoveryAuditDelay = null) { var coordinator = kit.CreateTestProbe(); var mux = kit.CreateTestProbe(); @@ -159,7 +163,8 @@ public static class V3ArtifactSeeder dependencyMux: mux.Ref, opcUaPublishActor: opcUaPublishActor, virtualTagEvaluator: NullVirtualTagEvaluator.Instance, - virtualTagHostOverride: vtHost.Ref)); + virtualTagHostOverride: vtHost.Ref, + commsRecoveryAuditDelay: commsRecoveryAuditDelay)); return (actor, mux, coordinator); }