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 cab1515b..ed207cf7 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs @@ -50,9 +50,9 @@ namespace ZB.MOM.WW.OtOpcUa.Runtime.Drivers; /// maintains the FullName→NodeId live-value routing map, and forwards results to the /// OPC UA publish actor via ForwardBatchToMux — one whole driver production burst per message, /// and the ONLY value ingress from driver children ( wraps even a -/// single data change in an AttributeValueBatchPublished). The per-value -/// AttributeValuePublished survives only as the message -/// fabricates straight to the dependency mux; the host itself no longer receives it. +/// single data change in an AttributeValueBatchPublished). +/// fabricates a batch of the same shape straight to the dependency mux, so the host neither receives nor +/// sends the per-value AttributeValuePublished any more. /// public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedStash { @@ -284,6 +284,34 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS /// value-subscription set. private readonly Dictionary<(string DriverInstanceId, string RawPath), HashSet> _alarmNodeIdByDriverRef = new(); + /// + /// Secondary index over , grouping its entries by + /// DriverInstanceId: DriverInstanceId → [(RawPath, the SAME node set)]. Rebuilt whole in + /// the same pass that rebuilds the primary — never maintained + /// incrementally, so the two cannot drift. + /// + /// It exists because the connectivity handlers are the only readers that need one driver's + /// entries, and the primary is keyed by the composite (DriverInstanceId, RawPath) — so + /// answering "which refs does this driver own?" meant scanning every entry, i.e. the whole node's + /// address space, for a single driver's disconnect. On a 50k-tag node a driver flap walked 50k + /// keys to find its own few thousand. + /// + /// + /// The node set is held BY REFERENCE, not copied: the entry is captured when the primary's set is + /// created, so any node later added to that set (a UNS reference registered in the same pass) is + /// visible here without a second bookkeeping step. + /// + /// + private readonly Dictionary Nodes)>> _valueRefsByDriver = + new(StringComparer.Ordinal); + + /// The alarm-side twin of : a secondary index over + /// grouping its entries by DriverInstanceId, so the + /// connectivity handler annotates one driver's conditions without scanning every condition on the node. + /// Rebuilt whole in the same pass, and likewise holding each condition-node set by reference. + private readonly Dictionary Nodes)>> _alarmRefsByDriver = + new(StringComparer.Ordinal); + /// /// Inverse of : folder-scoped condition NodeId → /// (DriverInstanceId, FullName = alarm ConditionId/AlarmFullReference). Built in the SAME apply @@ -1231,16 +1259,20 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS var statusCode = msg.Connected ? OpcUaStatus.Good : OpcUaStatus.BadNoCommunication; var ts = DateTime.UtcNow; var annotated = 0; - // Fan out to every condition this driver owns. _alarmNodeIdByDriverRef is keyed by - // (DriverInstanceId, RawPath); one driver ref can back several condition NodeIds (identical machines). - foreach (var ((driverId, _), nodeIds) in _alarmNodeIdByDriverRef) + // Fan out to every condition this driver owns, resolved through the per-driver index rather than by + // scanning every condition on the node: _alarmNodeIdByDriverRef is keyed by the composite + // (DriverInstanceId, RawPath), so a filtered walk of it costs the whole address space per flap. One + // driver ref can still back several condition NodeIds (identical machines) — hence the inner loop. + if (_alarmRefsByDriver.TryGetValue(msg.DriverInstanceId, out var alarmEntries)) { - if (driverId != msg.DriverInstanceId) continue; - foreach (var n in nodeIds) + foreach (var (_, nodeIds) in alarmEntries) { - _opcUaPublishActor.Tell(new ZB.MOM.WW.OtOpcUa.Runtime.OpcUa.OpcUaPublishActor.AlarmQualityUpdate( - n.NodeId, statusCode, ts, n.Realm)); - annotated++; + foreach (var n in nodeIds) + { + _opcUaPublishActor.Tell(new ZB.MOM.WW.OtOpcUa.Runtime.OpcUa.OpcUaPublishActor.AlarmQualityUpdate( + n.NodeId, statusCode, ts, n.Realm)); + annotated++; + } } } @@ -1258,7 +1290,9 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS /// /// The ref set already existed. #481 asked for a driver → all-tag-refs map to be built; /// is exactly that, keyed (DriverInstanceId, RawPath) and - /// rebuilt from the composition's RawTags ∪ UnsReferenceVariables on every apply. + /// rebuilt from the composition's RawTags ∪ UnsReferenceVariables on every apply — + /// read here through the per-driver index, since that composite + /// key cannot answer "this driver's refs" without walking the node's whole address space. /// Alarm-bearing tags live in the separate and are /// deliberately absent here — a native condition is not a mux value, and its comms quality is /// annotated by the caller through the Part 9 path instead. @@ -1286,29 +1320,41 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS /// Ungated by redundancy role, for the same reason as the native-condition annotation above: a /// Secondary must keep its scripted-alarm evaluation coherent for failover. The mux fan-out /// reaches only refs something is actually subscribed to, so a driver whose tags feed no script - /// and no virtual tag costs one dictionary scan and no delivery. + /// and no virtual tag costs one dictionary lookup, one envelope, and no delivery. + /// + /// + /// ONE batch, not one message per ref. The mux's batch receive filters per item against the + /// same registered-interest map its per-value receive consulted, so what each subscriber is handed + /// is unchanged — but a driver owning 5k tags now costs one Tell on its disconnect edge instead of + /// 5k, on the actor that also carries the node's steady value ingress. /// /// /// The driver child that just reported comms loss. private void FanCommsLossToDependencyMux(string driverInstanceId) { if (_dependencyMux is null) return; + // Indexed, not a filtered scan of the composite-keyed primary: this used to walk every value ref on + // the node to find one driver's. + if (!_valueRefsByDriver.TryGetValue(driverInstanceId, out var entries) || entries.Count == 0) return; var ts = DateTime.UtcNow; - var pushed = 0; - foreach (var (driverId, rawPath) in _nodeIdByDriverRef.Keys) + var pushed = entries.Count; + // ONE envelope for the whole driver, not one Tell per ref: a driver owning 5k tags cost 5k messages + // on a single connectivity edge. The mux's batch path filters the items per registered interest + // exactly as it does a steady production burst, so the delivery is unchanged — only the framing is. + var items = new DriverInstanceActor.AttributeValueItem[pushed]; + for (var i = 0; i < pushed; i++) { - if (driverId != driverInstanceId) continue; - - _dependencyMux.Tell(new DriverInstanceActor.AttributeValuePublished( - driverInstanceId, rawPath, Value: null, OpcUaStatus.BadNoCommunication, ts)); - pushed++; + items[i] = new DriverInstanceActor.AttributeValueItem( + entries[i].RawPath, Value: null, OpcUaStatus.BadNoCommunication, ts); } - if (pushed > 0) - _log.Debug( - "DriverHost {Node}: driver {Driver} disconnected — pushed BadNoCommunication for {Count} mux ref(s)", - _localNode, driverInstanceId, pushed); + _dependencyMux.Tell(new DriverInstanceActor.AttributeValueBatchPublished(driverInstanceId, items)); + + // Unconditional: the empty case returned above, so reaching here means at least one ref was pushed. + _log.Debug( + "DriverHost {Node}: driver {Driver} disconnected — pushed BadNoCommunication for {Count} mux ref(s) in one batch", + _localNode, driverInstanceId, pushed); } private void ForwardNativeAlarm(DriverInstanceActor.AttributeAlarmPublished msg) @@ -2455,6 +2501,18 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS ApplyDesiredSubscriptions(composition); } + /// Gets (creating on first use) the per-driver bucket of a secondary index. + /// Either or . + /// The driver whose bucket is wanted. + /// The mutable bucket for that driver. + private static List<(string RawPath, HashSet Nodes)> IndexFor( + Dictionary Nodes)>> index, string driverInstanceId) + { + if (!index.TryGetValue(driverInstanceId, out var bucket)) + index[driverInstanceId] = bucket = new List<(string, HashSet)>(); + return bucket; + } + /// /// Hand each running driver child its desired value + alarm subscription set for /// , rebuild the live-value/write routing maps, and forward the @@ -2509,6 +2567,10 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS _driverRefByAlarmNodeId.Clear(); _alarmMetaByNodeId.Clear(); _nativeAlarmProjector.Clear(); + // The per-driver secondary indexes are rebuilt WHOLE alongside their primaries below — same pass, + // same clear — so no incremental maintenance exists to get out of step with them. + _valueRefsByDriver.Clear(); + _alarmRefsByDriver.Clear(); foreach (var t in composition.RawTags) { var key = (t.DriverInstanceId, t.NodeId); // (DriverInstanceId, RawPath) — RawPath IS the wire-ref @@ -2517,7 +2579,13 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS // Native alarm → a single Part 9 condition at the RawPath (Raw realm). WP4 adds the referencing // equipment notifier NodeIds; WP3 wires the single raw condition. if (!_alarmNodeIdByDriverRef.TryGetValue(key, out var aset)) + { _alarmNodeIdByDriverRef[key] = aset = new HashSet(); + // Indexed on CREATION only, so a repeated key contributes exactly one entry — and the set + // is captured by reference, so later additions to it are seen through the index. + IndexFor(_alarmRefsByDriver, t.DriverInstanceId).Add((t.NodeId, aset)); + } + aset.Add(new NodeRealmRef(t.NodeId, AddressSpaceRealm.Raw)); // Inverse 1:1 map for the inbound native-condition ack path (keyed by BARE condition NodeId). _driverRefByAlarmNodeId[t.NodeId] = key; @@ -2532,7 +2600,11 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS // Value tag: register the RAW NodeId (Raw realm) AND every referencing UNS NodeId (Uns realm) // against the SAME driver ref — the single value source fans to all of them. if (!_nodeIdByDriverRef.TryGetValue(key, out var set)) + { _nodeIdByDriverRef[key] = set = new HashSet(); + IndexFor(_valueRefsByDriver, t.DriverInstanceId).Add((t.NodeId, set)); + } + set.Add(new NodeRealmRef(t.NodeId, AddressSpaceRealm.Raw)); _driverRefByNodeId[(AddressSpaceRealm.Raw, t.NodeId)] = key; diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorCommsLossTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorCommsLossTests.cs new file mode 100644 index 00000000..c1c1854e --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorCommsLossTests.cs @@ -0,0 +1,130 @@ +using Akka.Actor; +using Akka.TestKit; +using Microsoft.EntityFrameworkCore; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Commons.Engines; +using ZB.MOM.WW.OtOpcUa.Commons.Messages.Deploy; +using ZB.MOM.WW.OtOpcUa.Commons.OpcUa; +using ZB.MOM.WW.OtOpcUa.Commons.Types; +using ZB.MOM.WW.OtOpcUa.Configuration; +using ZB.MOM.WW.OtOpcUa.Runtime.Drivers; +using ZB.MOM.WW.OtOpcUa.Runtime.VirtualTags; +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 comms-loss fan (DriverHostActor.FanCommsLossToDependencyMux) end-to-end through a REAL +/// , rather than the mux probe the other host suites use. +/// +/// Two properties are pinned here that a probe cannot show. First, the fan is ONE +/// for the whole driver rather than one +/// Tell per ref — a driver owning 5k tags used to cost 5k messages on a single connectivity edge — and +/// the mux's batch path still delivers every item to an interested subscriber, so batching did not +/// quietly drop the delivery it exists to make. +/// +/// +/// Second, the driver scoping: the fan resolves its refs through a per-driver index built alongside +/// _nodeIdByDriverRef instead of scanning every entry in it, and an index is exactly the thing +/// that can key a lookup wrong. The two-driver case is therefore the falsification control — the +/// untouched driver's ref must be absent from the batch, which a scan-and-filter implementation and a +/// correctly-keyed index both satisfy but a mis-keyed one does not. +/// +/// +public sealed class DriverHostActorCommsLossTests : RuntimeActorTestBase +{ + private static readonly NodeId TestNode = NodeId.Parse("driver-commsloss-test"); + private static readonly RevisionHash RevA = RevisionHash.Parse(new string('c', 64)); + + /// Spawns a host whose dependency mux is a REAL , applies the + /// seeded deployment, and registers 's interest in . + /// + /// The factory whose deployment was seeded by . + /// The deployment to dispatch. + /// The probe standing in for a VirtualTagActor / scripted-alarm subscriber. + /// The RawPaths that subscriber registers interest in. + /// The applied host actor. + private IActorRef SpawnHostWithRealMux( + IDbContextFactory db, DeploymentId deploymentId, TestProbe subscriber, params string[] refs) + { + var coordinator = CreateTestProbe(); + var mux = Sys.ActorOf(DependencyMuxActor.Props()); + mux.Tell(new DependencyMuxActor.RegisterInterest(refs, subscriber.Ref)); + + var actor = Sys.ActorOf(DriverHostActor.Props( + db, TestNode, coordinator.Ref, + localRoles: new HashSet { "driver" }, + dependencyMux: mux, + virtualTagEvaluator: NullVirtualTagEvaluator.Instance, + virtualTagHostOverride: CreateTestProbe().Ref)); + + actor.Tell(new DispatchDeployment(deploymentId, RevA, CorrelationId.NewId())); + coordinator.ExpectMsg(PresenceBudget).Outcome.ShouldBe(ApplyAckOutcome.Applied); + + return actor; + } + + /// A driver's comms loss reaches the mux as ONE batch carrying a Bad item per ref it owns, and + /// the mux delivers that whole slice to an interested subscriber in ONE + /// . + [Fact] + public void Comms_loss_reaches_the_mux_as_one_batch_of_bad_items() + { + var db = NewInMemoryDbFactory(); + var deploymentId = SeedV3Deployment(db, RevA, + rawTags: new[] + { + (Driver: "drv-1", DriverName: "Modbus", Device: "dev1", Tag: "speed", DataType: "Double", Writable: false), + (Driver: "drv-1", DriverName: "Modbus", Device: "dev1", Tag: "temp", DataType: "Double", Writable: false), + }, + unsRefs: Array.Empty<(string, string, string, string, string?)>()); + + var subscriber = CreateTestProbe(); + var actor = SpawnHostWithRealMux(db, deploymentId, subscriber, + "Plant/Modbus/dev1/speed", "Plant/Modbus/dev1/temp"); + + actor.Tell(new DriverInstanceActor.ConnectivityChanged("drv-1", Connected: false)); + + var batch = subscriber.ExpectMsg(PresenceBudget); + batch.Items.Select(i => i.TagId).ShouldBe( + new[] { "Plant/Modbus/dev1/speed", "Plant/Modbus/dev1/temp" }, ignoreOrder: true); + foreach (var item in batch.Items) + { + item.StatusCode.ShouldBe(OpcUaStatus.BadNoCommunication); + item.Value.ShouldBeNull(); + } + + // ONE batch, not one per ref: the whole point of the change. A per-ref fan would land a second + // single-item DependencyValueBatch here. + subscriber.ExpectNoMsg(TimeSpan.FromMilliseconds(300)); + } + + /// Falsification control for the per-driver index: with TWO drivers configured, one driver's + /// comms loss carries ONLY its own refs — the other driver's tag is untouched. + [Fact] + public void Comms_loss_batch_carries_only_the_dead_drivers_refs() + { + var db = NewInMemoryDbFactory(); + var deploymentId = SeedV3Deployment(db, RevA, + rawTags: new[] + { + (Driver: "drv-1", DriverName: "Modbus", Device: "dev1", Tag: "speed", DataType: "Double", Writable: false), + (Driver: "drv-1", DriverName: "Modbus", Device: "dev1", Tag: "temp", DataType: "Double", Writable: false), + (Driver: "drv-2", DriverName: "Modbus2", Device: "dev2", Tag: "flow", DataType: "Double", Writable: false), + }, + unsRefs: Array.Empty<(string, string, string, string, string?)>()); + + var subscriber = CreateTestProbe(); + var actor = SpawnHostWithRealMux(db, deploymentId, subscriber, + "Plant/Modbus/dev1/speed", "Plant/Modbus/dev1/temp", "Plant/Modbus2/dev2/flow"); + + actor.Tell(new DriverInstanceActor.ConnectivityChanged("drv-1", Connected: false)); + + var batch = subscriber.ExpectMsg(PresenceBudget); + batch.Items.Select(i => i.TagId).ShouldBe( + new[] { "Plant/Modbus/dev1/speed", "Plant/Modbus/dev1/temp" }, ignoreOrder: true); + subscriber.ExpectNoMsg(TimeSpan.FromMilliseconds(300)); + } +} diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorLiveValueTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorLiveValueTests.cs index d86e4ce9..7a3a0052 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorLiveValueTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorLiveValueTests.cs @@ -160,7 +160,8 @@ public sealed class DriverHostActorLiveValueTests : RuntimeActorTestBase // went on evaluating it. These pin the synthetic Bad that closes that hole. /// Comms loss pushes a Bad for every value ref the driver owns — not one, and not the - /// alarm-bearing ones (those take the Part 9 quality path instead). + /// alarm-bearing ones (those take the Part 9 quality path instead) — as ONE batch envelope rather than + /// one Tell per ref. [Fact] public void Driver_comms_loss_pushes_BadNoCommunication_into_the_mux_for_every_ref_it_owns() { @@ -177,11 +178,10 @@ public sealed class DriverHostActorLiveValueTests : RuntimeActorTestBase actor.Tell(new DriverInstanceActor.ConnectivityChanged("drv-1", Connected: false)); - var pushes = new[] - { - mux.ExpectMsg(PresenceBudget), - mux.ExpectMsg(PresenceBudget), - }; + // ONE envelope for the whole driver: a driver owning 5k tags cost 5k Tells before this was batched. + var fan = mux.ExpectMsg(PresenceBudget); + fan.DriverInstanceId.ShouldBe("drv-1"); + var pushes = fan.Items; pushes.Select(p => p.FullReference).ShouldBe( new[] { "Plant/Modbus/dev1/speed", "Plant/Modbus/dev1/temp" }, ignoreOrder: true); @@ -194,7 +194,6 @@ public sealed class DriverHostActorLiveValueTests : RuntimeActorTestBase // Null rather than the last-known value: the honest statement is "no current value". The // engine's AreInputsReady gate holds the predicate on any Bad, so no script sees the null. p.Value.ShouldBeNull(); - p.DriverInstanceId.ShouldBe("drv-1"); } mux.ExpectNoMsg(TimeSpan.FromMilliseconds(300));