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 42222d6d..e56b1c31 100644
--- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs
+++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs
@@ -48,8 +48,11 @@ 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 (per value) and ForwardBatchToMux (one whole
-/// driver production burst per message — the primary path for a batch-capable driver).
+/// 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.
///
public sealed class DriverHostActor : ReceiveActor, IWithTimers
{
@@ -186,7 +189,7 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers
///
/// Driver live-value routing map: (DriverInstanceId, RawPath) → materialised NodeId(s) (realm-tagged).
/// Rebuilt every apply by from the composition's RawTags
- /// ∪ UnsReferenceVariables, and resolved in so a driver value
+ /// ∪ UnsReferenceVariables, and resolved in so a driver value
/// published by wire-ref RawPath fans to its raw node AND every referencing UNS node with identical
/// value/quality/timestamp — the single-source-fan-out (no independent buffer, so no drift). A set
/// because one driver ref can back the raw node plus N UNS references; the per-apply rebuild dedups.
@@ -924,7 +927,6 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers
{
Receive(HandleDispatchFromSteady);
Receive(HandleGetDiagnostics);
- Receive(ForwardToMux);
Receive(ForwardBatchToMux);
Receive(ForwardNativeAlarm);
Receive(OnDriverConnectivityChanged);
@@ -959,7 +961,6 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers
// then Become(Applying)); completing the apply transitions back to Steady.
Receive(HandleFetchedForApply);
Receive(HandleGetDiagnostics);
- Receive(ForwardToMux);
Receive(ForwardBatchToMux);
Receive(ForwardNativeAlarm);
Receive(OnDriverConnectivityChanged);
@@ -977,16 +978,25 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers
}
///
- /// Routes one batched driver production burst — the batched analogue of
- /// , and the primary value ingress for a batch-capable driver.
+ /// Routes one batched driver production burst — the ONLY value ingress from a driver child, since
+ /// wraps even a single data change in an
+ /// .
///
- /// 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
+ /// Each item's FullReference (the tag's RawPath) resolves through
+ /// — rebuilt every apply from the composition's RawTags
+ /// ∪ UnsReferenceVariables — to the v3 Batch 4 single-source fan-out: the RAW
+ /// NodeId () AND every referencing UNS NodeId
+ /// (), each carrying IDENTICAL value/status/timestamp, so
+ /// there is no independent buffer to drift. An unmapped ref is Debug-logged and dropped without
+ /// disturbing the rest of the burst.
+ ///
+ ///
+ /// The whole burst costs ONE hop to the dependency mux (which filters per item itself, feeding
+ /// VirtualTag inputs keyed by RawPath) and ONE
/// to
- /// the publish actor, instead of one message per value.
+ /// the publish actor — never one message per value. Both sinks are optional: without a mux,
+ /// VirtualTagActor evaluation can't fire (the dev/Mac path, no virtual tags registered);
+ /// production binds both via the RuntimeActors extension.
///
///
private void ForwardBatchToMux(DriverInstanceActor.AttributeValueBatchPublished msg)
@@ -998,13 +1008,17 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers
// 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.
+ var count = msg.Items.Count;
List? writes = null;
- for (var idx = 0; idx < msg.Items.Count; idx++)
+ for (var idx = 0; idx < count; idx++)
{
var item = msg.Items[idx];
if (_nodeIdByDriverRef.TryGetValue((msg.DriverInstanceId, item.FullReference), out var nodeIds))
{
- writes ??= new List(msg.Items.Count * 2);
+ // Sized for the dominant case — every item resolving to its raw node alone, no UNS
+ // reference — where this is exact and the list never grows. A burst that does fan out
+ // pays at most a doubling or two, which beats over-allocating on every burst that doesn't.
+ writes ??= new List(count);
foreach (var n in nodeIds)
writes.Add(new ValueWrite(n.NodeId, item.Value, item.StatusCode, item.TimestampUtc, n.Realm));
}
@@ -1020,42 +1034,10 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers
_opcUaPublishActor.Tell(new ZB.MOM.WW.OtOpcUa.Runtime.OpcUa.OpcUaPublishActor.AttributeValueBatchUpdate(writes));
}
- private void ForwardToMux(DriverInstanceActor.AttributeValuePublished msg)
- {
- // Pass driver-published values to the dependency mux when one is wired (VirtualTag inputs,
- // keyed by FullReference). Without a mux, VirtualTagActor evaluation can't fire — that's the
- // dev/Mac path (no virtual tags registered); production binds the mux via the RuntimeActors
- // extension. KEEP this unchanged — VirtualTag inputs are still keyed by FullReference.
- _dependencyMux?.Tell(msg);
-
- if (_opcUaPublishActor is null) return;
-
- // Route the value to the OPC UA sink at the variable's ACTUAL NodeId. Equipment-tag variables
- // are materialised with folder-scoped NodeIds (EquipmentId/FolderPath/Name), while the driver
- // publishes keyed by its wire-ref FullName (FullReference). The _nodeIdByDriverRef map — built
- // each apply from the composition's EquipmentTags — resolves (DriverInstanceId, FullName) to
- // the folder-scoped NodeId(s) the materialiser placed the variable(s) at, so the value lands
- // instead of leaving the variable at BadWaitingForInitialData. One driver ref can back several
- // equipment variables (identical machines sharing a register), hence the fan-out.
- if (_nodeIdByDriverRef.TryGetValue((msg.DriverInstanceId, msg.FullReference), out var nodeIds))
- {
- // v3 Batch 4 single-source fan-out: one driver publish for a RawPath lands on the raw NodeId AND
- // every referencing UNS NodeId, each with its realm, carrying IDENTICAL value/status/timestamp.
- foreach (var n in nodeIds)
- _opcUaPublishActor.Tell(new ZB.MOM.WW.OtOpcUa.Runtime.OpcUa.OpcUaPublishActor.AttributeValueUpdate(
- n.NodeId, msg.Value, msg.StatusCode, msg.TimestampUtc, n.Realm));
- }
- else
- {
- _log.Debug("DriverHost {Node}: no bound NodeId for ({Driver},{Ref}) — value dropped",
- _localNode, msg.DriverInstanceId, msg.FullReference);
- }
- }
-
///
/// Routes a native alarm transition (published by a driver child as
/// ) to its materialised Part 9 condition
- /// node(s). The alarm path analogue of : the transition's ConditionId
+ /// node(s). The alarm path analogue of : the transition's ConditionId
/// (the dotted alarm full-reference, which equals the authored equipment-tag FullName — NOT the bare
/// SourceNodeId owning object) is resolved by the map —
/// built each apply from the alarm-bearing EquipmentTags — to the folder-scoped condition NodeId(s)
diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/VirtualTags/DependencyMuxActor.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/VirtualTags/DependencyMuxActor.cs
index 69fb5bb4..cf06b01f 100644
--- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/VirtualTags/DependencyMuxActor.cs
+++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/VirtualTags/DependencyMuxActor.cs
@@ -13,11 +13,12 @@ namespace ZB.MOM.WW.OtOpcUa.Runtime.VirtualTags;
/// subscriber.
///
/// DriverHostActor forwards driver-child value publishes to this mux (one mux per driver-role
-/// node). The PRIMARY ingress is the batched
+/// node). The ONLY driver ingress is the batched
/// — one envelope per driver
-/// production burst, which the mux filters per item — while the per-value
+/// production burst, which the mux filters per item. The per-value
/// receive stays for the comms-loss fan
-/// (DriverHostActor.FanCommsLossToDependencyMux fabricates it) and per-tag drivers. Both
+/// ALONE: DriverHostActor.FanCommsLossToDependencyMux fabricates that message and Tells this
+/// mux directly, never through the host's own receive. Both
/// produce the same per interested subscriber.
/// The mux is deliberately not a DPS subscriber — virtual-tag evaluation is local to each node and
/// would over-emit if it spanned the cluster.
@@ -116,12 +117,19 @@ public sealed class DependencyMuxActor : ReceiveActor
/// Filters one batched driver production burst per item — the batched analogue of
/// . Items nobody registered for are dropped (the common case:
/// only a fraction of an address space feeds virtual-tag expressions), so a burst of N values costs one
- /// message plus one DependencyValueChanged per genuinely interested (item, subscriber) pair.
+ /// message plus one DependencyValueChanged per genuinely interested (item, subscriber) pair.
+ /// The registered set is NOT just virtual-tag inputs: ContinuousHistorizationRecorder
+ /// registers interest in every historized ref, so on a historizing node can be
+ /// large and a sizeable fraction of each burst resolves.
private void OnAttributeValueBatchPublished(DriverInstanceActor.AttributeValueBatchPublished msg)
{
+ // Nobody has registered anything — skip the whole burst rather than probing an empty map per item.
+ if (_byRef.Count == 0) return;
+
// Indexed for-loop, not foreach: this is the hot value ingress and an IReadOnlyList enumerator
// would allocate once per burst.
- for (var i = 0; i < msg.Items.Count; i++)
+ var count = msg.Items.Count;
+ for (var i = 0; i < count; i++)
{
var item = msg.Items[i];
if (!_byRef.TryGetValue(item.FullReference, out var subscribers) || subscribers.Count == 0) continue;
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
index 8205e02b..51e3ee87 100644
--- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorBatchFanOutTests.cs
+++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorBatchFanOutTests.cs
@@ -1,18 +1,12 @@
-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;
+using static ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness.V3ArtifactSeeder;
namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Drivers;
@@ -29,7 +23,8 @@ namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Drivers;
///
///
/// Arrangement mirrors — a real apply through the seeded
-/// v3 artifact builds the map; sink + mux are s.
+/// v3 artifact () builds the map; sink + mux are
+/// s.
///
///
public sealed class DriverHostActorBatchFanOutTests : RuntimeActorTestBase
@@ -54,7 +49,7 @@ public sealed class DriverHostActorBatchFanOutTests : RuntimeActorTestBase
},
unsRefs: new[] { (Area: "filling", Line: "line1", Equip: "station1", BackingTag: "speed", Effective: (string?)null) });
- var (actor, publish, _) = SpawnHostAndApply(db, deploymentId);
+ var (actor, publish, _) = SpawnHostAndApply(this, db, TestNode, deploymentId, RevA);
actor.Tell(new DriverInstanceActor.AttributeValueBatchPublished("drv-1", new[]
{
@@ -62,7 +57,7 @@ public sealed class DriverHostActorBatchFanOutTests : RuntimeActorTestBase
new DriverInstanceActor.AttributeValueItem("Plant/Modbus/dev1/temp", 7.5, OpcUaStatus.BadNoCommunication, TsB),
}));
- var batch = publish.ExpectMsg(TimeSpan.FromSeconds(5));
+ var batch = publish.ExpectMsg(PresenceBudget);
batch.Items.Count.ShouldBe(3);
var byId = batch.Items.ToDictionary(w => w.NodeId);
@@ -103,7 +98,7 @@ public sealed class DriverHostActorBatchFanOutTests : RuntimeActorTestBase
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);
+ var (actor, publish, _) = SpawnHostAndApply(this, db, TestNode, deploymentId, RevA);
actor.Tell(new DriverInstanceActor.AttributeValueBatchPublished("drv-1", new[]
{
@@ -111,7 +106,7 @@ public sealed class DriverHostActorBatchFanOutTests : RuntimeActorTestBase
new DriverInstanceActor.AttributeValueItem("Plant/Modbus/dev1/speed", 42.0, OpcUaStatus.Good, TsA),
}));
- var batch = publish.ExpectMsg(TimeSpan.FromSeconds(5));
+ var batch = publish.ExpectMsg(PresenceBudget);
batch.Items.Count.ShouldBe(1);
batch.Items[0].NodeId.ShouldBe("Plant/Modbus/dev1/speed");
batch.Items[0].Value.ShouldBe(42.0);
@@ -121,7 +116,7 @@ public sealed class DriverHostActorBatchFanOutTests : RuntimeActorTestBase
{
new DriverInstanceActor.AttributeValueItem("Plant/Modbus/dev1/speed", 43.0, OpcUaStatus.Good, TsB),
}));
- publish.ExpectMsg(TimeSpan.FromSeconds(5))
+ publish.ExpectMsg(PresenceBudget)
.Items[0].Value.ShouldBe(43.0);
}
@@ -139,7 +134,7 @@ public sealed class DriverHostActorBatchFanOutTests : RuntimeActorTestBase
},
unsRefs: Array.Empty<(string, string, string, string, string?)>());
- var (actor, _, mux) = SpawnHostAndApply(db, deploymentId);
+ var (actor, _, mux) = SpawnHostAndApply(this, db, TestNode, deploymentId, RevA);
actor.Tell(new DriverInstanceActor.AttributeValueBatchPublished("drv-1", new[]
{
@@ -147,7 +142,7 @@ public sealed class DriverHostActorBatchFanOutTests : RuntimeActorTestBase
new DriverInstanceActor.AttributeValueItem("Plant/Modbus/dev1/temp", 7.5, OpcUaStatus.Good, TsB),
}));
- var fwd = mux.ExpectMsg(TimeSpan.FromSeconds(5));
+ var fwd = mux.ExpectMsg(PresenceBudget);
fwd.DriverInstanceId.ShouldBe("drv-1");
fwd.Items.Select(i => i.FullReference).ShouldBe(
new[] { "Plant/Modbus/dev1/speed", "Plant/Modbus/dev1/temp" });
@@ -165,7 +160,7 @@ public sealed class DriverHostActorBatchFanOutTests : RuntimeActorTestBase
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);
+ var (actor, publish, mux) = SpawnHostAndApply(this, db, TestNode, deploymentId, RevA);
actor.Tell(new DriverInstanceActor.AttributeValueBatchPublished("drv-1", new[]
{
@@ -173,104 +168,31 @@ public sealed class DriverHostActorBatchFanOutTests : RuntimeActorTestBase
new DriverInstanceActor.AttributeValueItem("Plant/Modbus/dev1/alsonope", 2.0, OpcUaStatus.Good, TsB),
}));
- mux.ExpectMsg(TimeSpan.FromSeconds(5))
+ mux.ExpectMsg(PresenceBudget)
.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)
+ /// A host with NO OPC UA publish actor wired (the ctor default — the dev/Mac shape) still
+ /// makes the mux hop: the batch fan-out returns early on a null sink, and returning early BEFORE the
+ /// mux Tell would silently strand virtual-tag and continuous-historization inputs.
+ [Fact]
+ public void Mux_still_receives_batch_when_no_publish_actor_is_wired()
{
- var coordinator = CreateTestProbe();
- var publish = CreateTestProbe();
- var mux = CreateTestProbe();
- var vtHost = CreateTestProbe();
+ 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 = Sys.ActorOf(DriverHostActor.Props(
- db, TestNode, coordinator.Ref,
- localRoles: new HashSet { "driver" },
- dependencyMux: mux.Ref,
- opcUaPublishActor: publish.Ref,
- virtualTagEvaluator: NullVirtualTagEvaluator.Instance,
- virtualTagHostOverride: vtHost.Ref));
+ var (actor, mux) = SpawnHostAndApplyWithoutPublishActor(this, db, TestNode, deploymentId, RevA);
- 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
+ actor.Tell(new DriverInstanceActor.AttributeValueBatchPublished("drv-1", 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();
+ new DriverInstanceActor.AttributeValueItem("Plant/Modbus/dev1/speed", 42.0, OpcUaStatus.Good, TsA),
+ }));
- 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