From c51568f2a5a4e05e35f08fa7356021c6089a8840 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Sat, 15 Aug 2026 12:36:01 -0400 Subject: [PATCH] feat(runtime): DriverHostActor batched value fan-out MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replaces the temporary per-item unwrap bridge with a real batched handler: one AttributeValueBatchPublished burst now costs ONE hop to the dependency mux and ONE OpcUaPublishActor.AttributeValueBatchUpdate, instead of one message per value. Routing per item is unchanged — the v3 Batch 4 single-source fan-out (raw NodeId + every referencing UNS NodeId, identical value/status/timestamp), with an unmapped ref Debug-logged and dropped without disturbing the rest of the burst. A burst in which nothing resolves sends no publish message at all rather than an empty batch. Indexed for-loop over Items is deliberate: no interface-enumerator allocation per burst on the hottest routing seam. ForwardToMux and its registrations stay for the comms-loss fan and per-tag drivers. --- .../Drivers/DriverHostActor.cs | 49 +++- .../DriverHostActorBatchFanOutTests.cs | 276 ++++++++++++++++++ 2 files changed, 315 insertions(+), 10 deletions(-) create mode 100644 tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorBatchFanOutTests.cs 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 6310b9da..42222d6d 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs @@ -48,7 +48,8 @@ namespace ZB.MOM.WW.OtOpcUa.Runtime.Drivers; /// Children (DriverInstance/VirtualTag/etc.) are fully wired: DispatchDeployment drives /// which runs the spawn-plan via , /// maintains the FullName→NodeId live-value routing map, and forwards results to the -/// OPC UA publish actor via ForwardToMux. +/// OPC UA publish actor via ForwardToMux (per value) and ForwardBatchToMux (one whole +/// driver production burst per message — the primary path for a batch-capable driver). /// public sealed class DriverHostActor : ReceiveActor, IWithTimers { @@ -976,19 +977,47 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers } /// - /// Temporary bridge: unwrap a batched driver publish back into the per-value - /// path, so the batching ingress seam lands without changing any - /// downstream behaviour. + /// Routes one batched driver production burst — the batched analogue of + /// , and the primary value ingress for a batch-capable driver. + /// + /// The routing is identical per item: each FullReference (the tag's RawPath) resolves + /// through to the v3 Batch 4 single-source fan-out — the RAW + /// NodeId AND every referencing UNS NodeId, each carrying IDENTICAL value/status/timestamp — and + /// an unmapped ref is Debug-logged and dropped without disturbing the rest of the burst. What + /// changes is the delivery: the whole burst costs ONE hop to the dependency mux and ONE + /// to + /// the publish actor, instead of one message per value. + /// /// - // TODO(Task 8 plan 2026-08-15): replace with batched fan-out — one mux hop and one OPC UA publish - // hop per burst instead of one per value, which is the whole point of the batching work. - private void ForwardBatchToMux(DriverInstanceActor.AttributeValueBatchPublished batch) + private void ForwardBatchToMux(DriverInstanceActor.AttributeValueBatchPublished msg) { - foreach (var i in batch.Items) + // One envelope per burst; the mux filters per item (it registered a batch receive in Task 9). + _dependencyMux?.Tell(msg); + + if (_opcUaPublishActor is null) return; + + // Indexed for-loop, not foreach: IReadOnlyList enumerates through an interface + // enumerator that would allocate once per burst on the hottest routing seam in the server. + List? writes = null; + for (var idx = 0; idx < msg.Items.Count; idx++) { - ForwardToMux(new DriverInstanceActor.AttributeValuePublished( - batch.DriverInstanceId, i.FullReference, i.Value, i.StatusCode, i.TimestampUtc)); + var item = msg.Items[idx]; + if (_nodeIdByDriverRef.TryGetValue((msg.DriverInstanceId, item.FullReference), out var nodeIds)) + { + writes ??= new List(msg.Items.Count * 2); + foreach (var n in nodeIds) + writes.Add(new ValueWrite(n.NodeId, item.Value, item.StatusCode, item.TimestampUtc, n.Realm)); + } + else if (_log.IsDebugEnabled) + { + _log.Debug("DriverHost {Node}: no bound NodeId for ({Driver},{Ref}) — value dropped", + _localNode, msg.DriverInstanceId, item.FullReference); + } } + + // Nothing resolved ⇒ no message at all, rather than an empty batch for the sink to unpack. + if (writes is not null) + _opcUaPublishActor.Tell(new ZB.MOM.WW.OtOpcUa.Runtime.OpcUa.OpcUaPublishActor.AttributeValueBatchUpdate(writes)); } private void ForwardToMux(DriverInstanceActor.AttributeValuePublished msg) diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorBatchFanOutTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorBatchFanOutTests.cs new file mode 100644 index 00000000..8205e02b --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorBatchFanOutTests.cs @@ -0,0 +1,276 @@ +using System.Text.Json; +using Akka.Actor; +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.Configuration.Entities; +using ZB.MOM.WW.OtOpcUa.Configuration.Enums; +using ZB.MOM.WW.OtOpcUa.Runtime.Drivers; +using ZB.MOM.WW.OtOpcUa.Runtime.OpcUa; +using ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness; + +namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Drivers; + +/// +/// Batched value fan-out in : a driver child's +/// burst costs exactly ONE mux hop and +/// ONE OPC UA publish hop, regardless of how many values it carries — the point of the batching work. +/// +/// The routing itself is unchanged from the per-value path: each item's FullReference +/// (== the tag's RawPath) resolves through the host's _nodeIdByDriverRef map to the RAW NodeId +/// AND every referencing UNS NodeId, each carrying IDENTICAL value/status/timestamp (v3 Batch 4 +/// single-source fan-out), and an unmapped ref is dropped rather than faulting the burst. These pin +/// the message COUNT as well as the contents, because a per-item unwrap satisfies the contents alone. +/// +/// +/// Arrangement mirrors — a real apply through the seeded +/// v3 artifact builds the map; sink + mux are s. +/// +/// +public sealed class DriverHostActorBatchFanOutTests : RuntimeActorTestBase +{ + private static readonly NodeId TestNode = NodeId.Parse("driver-batch-test"); + private static readonly RevisionHash RevA = RevisionHash.Parse(new string('b', 64)); + private static readonly DateTime TsA = new(2026, 8, 15, 9, 0, 0, DateTimeKind.Utc); + private static readonly DateTime TsB = new(2026, 8, 15, 9, 0, 1, DateTimeKind.Utc); + + /// A two-item burst whose first item is referenced by one equipment lands as a SINGLE + /// AttributeValueBatchUpdate carrying all three (raw, uns, raw) writes verbatim — including a non-Good + /// sub-code (#519), which rides through unbucketed. + [Fact] + public void Batch_fans_out_to_all_node_realm_refs_in_one_publish_message() + { + 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: new[] { (Area: "filling", Line: "line1", Equip: "station1", BackingTag: "speed", Effective: (string?)null) }); + + var (actor, publish, _) = SpawnHostAndApply(db, deploymentId); + + actor.Tell(new DriverInstanceActor.AttributeValueBatchPublished("drv-1", new[] + { + new DriverInstanceActor.AttributeValueItem("Plant/Modbus/dev1/speed", 42.0, OpcUaStatus.Good, TsA), + new DriverInstanceActor.AttributeValueItem("Plant/Modbus/dev1/temp", 7.5, OpcUaStatus.BadNoCommunication, TsB), + })); + + var batch = publish.ExpectMsg(TimeSpan.FromSeconds(5)); + batch.Items.Count.ShouldBe(3); + + var byId = batch.Items.ToDictionary(w => w.NodeId); + byId.Keys.ShouldBe(new[] + { + "Plant/Modbus/dev1/speed", + "filling/line1/station1/speed", + "Plant/Modbus/dev1/temp", + }, ignoreOrder: true); + + // Item A: raw + UNS, identical value/quality/timestamp (fan-out drift guard). + foreach (var id in new[] { "Plant/Modbus/dev1/speed", "filling/line1/station1/speed" }) + { + byId[id].Value.ShouldBe(42.0); + byId[id].StatusCode.ShouldBe(OpcUaStatus.Good); + byId[id].SourceTimestampUtc.ShouldBe(TsA); + } + byId["Plant/Modbus/dev1/speed"].Realm.ShouldBe(AddressSpaceRealm.Raw); + byId["filling/line1/station1/speed"].Realm.ShouldBe(AddressSpaceRealm.Uns); + + // Item B: raw only, and its own status sub-code + timestamp — not item A's. + byId["Plant/Modbus/dev1/temp"].Value.ShouldBe(7.5); + byId["Plant/Modbus/dev1/temp"].StatusCode.ShouldBe(OpcUaStatus.BadNoCommunication); + byId["Plant/Modbus/dev1/temp"].SourceTimestampUtc.ShouldBe(TsB); + byId["Plant/Modbus/dev1/temp"].Realm.ShouldBe(AddressSpaceRealm.Raw); + + // ONE message, not one per value. + publish.ExpectNoMsg(TimeSpan.FromMilliseconds(300)); + } + + /// An item whose ref is bound to no NodeId is dropped; the rest of the burst still lands in the + /// one message and nothing faults. + [Fact] + public void Unmapped_items_are_dropped_not_faulted() + { + var db = NewInMemoryDbFactory(); + var deploymentId = SeedV3Deployment(db, RevA, + rawTags: new[] { (Driver: "drv-1", DriverName: "Modbus", Device: "dev1", Tag: "speed", DataType: "Double", Writable: false) }, + unsRefs: Array.Empty<(string, string, string, string, string?)>()); + + var (actor, publish, _) = SpawnHostAndApply(db, deploymentId); + + actor.Tell(new DriverInstanceActor.AttributeValueBatchPublished("drv-1", new[] + { + new DriverInstanceActor.AttributeValueItem("Plant/Modbus/dev1/nope", 99.0, OpcUaStatus.Good, TsA), + new DriverInstanceActor.AttributeValueItem("Plant/Modbus/dev1/speed", 42.0, OpcUaStatus.Good, TsA), + })); + + var batch = publish.ExpectMsg(TimeSpan.FromSeconds(5)); + batch.Items.Count.ShouldBe(1); + batch.Items[0].NodeId.ShouldBe("Plant/Modbus/dev1/speed"); + batch.Items[0].Value.ShouldBe(42.0); + + // The host is still alive and routing after the unmapped item. + actor.Tell(new DriverInstanceActor.AttributeValueBatchPublished("drv-1", new[] + { + new DriverInstanceActor.AttributeValueItem("Plant/Modbus/dev1/speed", 43.0, OpcUaStatus.Good, TsB), + })); + publish.ExpectMsg(TimeSpan.FromSeconds(5)) + .Items[0].Value.ShouldBe(43.0); + } + + /// The dependency mux gets ONE envelope carrying the whole burst — not N per-item messages. + /// The mux filters per item itself. + [Fact] + public void Batch_reaches_mux_once() + { + 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 (actor, _, mux) = SpawnHostAndApply(db, deploymentId); + + actor.Tell(new DriverInstanceActor.AttributeValueBatchPublished("drv-1", new[] + { + new DriverInstanceActor.AttributeValueItem("Plant/Modbus/dev1/speed", 42.0, OpcUaStatus.Good, TsA), + new DriverInstanceActor.AttributeValueItem("Plant/Modbus/dev1/temp", 7.5, OpcUaStatus.Good, TsB), + })); + + var fwd = mux.ExpectMsg(TimeSpan.FromSeconds(5)); + fwd.DriverInstanceId.ShouldBe("drv-1"); + fwd.Items.Select(i => i.FullReference).ShouldBe( + new[] { "Plant/Modbus/dev1/speed", "Plant/Modbus/dev1/temp" }); + mux.ExpectNoMsg(TimeSpan.FromMilliseconds(300)); + } + + /// A burst in which NOTHING resolves sends no publish message at all — an empty batch would be + /// wasted work at the sink — while the mux still gets its one envelope (virtual-tag inputs are keyed by + /// ref, independent of whether the ref is materialised as a node). + [Fact] + public void All_unmapped_batch_sends_nothing_to_publish_actor() + { + var db = NewInMemoryDbFactory(); + var deploymentId = SeedV3Deployment(db, RevA, + rawTags: new[] { (Driver: "drv-1", DriverName: "Modbus", Device: "dev1", Tag: "speed", DataType: "Double", Writable: false) }, + unsRefs: Array.Empty<(string, string, string, string, string?)>()); + + var (actor, publish, mux) = SpawnHostAndApply(db, deploymentId); + + actor.Tell(new DriverInstanceActor.AttributeValueBatchPublished("drv-1", new[] + { + new DriverInstanceActor.AttributeValueItem("Plant/Modbus/dev1/nope", 1.0, OpcUaStatus.Good, TsA), + new DriverInstanceActor.AttributeValueItem("Plant/Modbus/dev1/alsonope", 2.0, OpcUaStatus.Good, TsB), + })); + + mux.ExpectMsg(TimeSpan.FromSeconds(5)) + .Items.Count.ShouldBe(2); + publish.ExpectNoMsg(TimeSpan.FromMilliseconds(500)); + } + + /// Spawns the host with publish + mux probes, dispatches the deployment, and waits for the + /// Applied ACK so the map build in PushDesiredSubscriptions has completed before the test publishes. + private (IActorRef Actor, Akka.TestKit.TestProbe Publish, Akka.TestKit.TestProbe Mux) SpawnHostAndApply( + IDbContextFactory db, DeploymentId deploymentId) + { + var coordinator = CreateTestProbe(); + var publish = CreateTestProbe(); + var mux = CreateTestProbe(); + var vtHost = CreateTestProbe(); + + var actor = Sys.ActorOf(DriverHostActor.Props( + db, TestNode, coordinator.Ref, + localRoles: new HashSet { "driver" }, + dependencyMux: mux.Ref, + opcUaPublishActor: publish.Ref, + virtualTagEvaluator: NullVirtualTagEvaluator.Instance, + virtualTagHostOverride: vtHost.Ref)); + + actor.Tell(new DispatchDeployment(deploymentId, RevA, CorrelationId.NewId())); + coordinator.ExpectMsg(TimeSpan.FromSeconds(5)).Outcome.ShouldBe(ApplyAckOutcome.Applied); + publish.ExpectMsg(TimeSpan.FromSeconds(5)); + + return (actor, publish, mux); + } + + /// Seeds a Sealed deployment whose artifact carries the v3 raw-tag chain (RawFolder "Plant" → + /// DriverInstance(RawFolderId) → Device → Tag) plus optional UnsTagReferences projecting a raw tag into an + /// Area/Line/Equipment — the same shape seeds. + private 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) + { + 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 }) + .ToArray(); + var devices = rawTags + .Select(t => (t.Driver, t.Device)) + .Distinct() + .Select(d => new { DeviceId = $"{d.Driver}:{d.Device}", DriverInstanceId = d.Driver, Name = d.Device, DeviceConfig = "{}" }) + .ToArray(); + var tags = rawTags.Select(t => new + { + TagId = t.Tag, + DeviceId = $"{t.Driver}:{t.Device}", + TagGroupId = (string?)null, + Name = t.Tag, + DataType = t.DataType, + AccessLevel = t.Writable ? 1 : 0, + TagConfig = "{}", + }).ToArray(); + + var areas = unsRefs.Select(r => r.Area).Distinct(StringComparer.Ordinal) + .Select(a => new { UnsAreaId = $"a-{a}", Name = a, ClusterId = "c1" }).ToArray(); + var lines = unsRefs.Select(r => (r.Area, r.Line)).Distinct() + .Select(l => new { UnsLineId = $"l-{l.Area}-{l.Line}", UnsAreaId = $"a-{l.Area}", Name = l.Line }).ToArray(); + var equipment = unsRefs.Select(r => (r.Area, r.Line, r.Equip)).Distinct() + .Select(e => new { EquipmentId = $"EQ-{e.Area}-{e.Line}-{e.Equip}", UnsLineId = $"l-{e.Area}-{e.Line}", Name = e.Equip, MachineCode = e.Equip }).ToArray(); + var unsTagReferences = unsRefs.Select((r, i) => new + { + UnsTagReferenceId = $"ref-{i}", + EquipmentId = $"EQ-{r.Area}-{r.Line}-{r.Equip}", + TagId = r.BackingTag, + DisplayNameOverride = r.Effective, + }).ToArray(); + + var artifact = JsonSerializer.SerializeToUtf8Bytes(new + { + RawFolders = new[] { new { RawFolderId = "rf-plant", ParentRawFolderId = (string?)null, Name = "Plant", ClusterId = "c1" } }, + DriverInstances = drivers, + Devices = devices, + TagGroups = Array.Empty(), + Tags = tags, + UnsAreas = areas, + UnsLines = lines, + Equipment = equipment, + UnsTagReferences = unsTagReferences, + }); + + var id = DeploymentId.NewId(); + using var ctx = db.CreateDbContext(); + ctx.Deployments.Add(new Deployment + { + DeploymentId = id.Value, + RevisionHash = rev.Value, + Status = DeploymentStatus.Sealed, + CreatedBy = "test", + SealedAtUtc = DateTime.UtcNow, + ArtifactBlob = artifact, + }); + ctx.SaveChanges(); + return id; + } +}