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