From 8ef02506f7a1031443cd83e2a3b137bc36c99793 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Sat, 15 Aug 2026 12:31:29 -0400 Subject: [PATCH] =?UTF-8?q?fix(runtime):=20harden=20batch=20forwarding=20?= =?UTF-8?q?=E2=80=94=20detach=20pins,=20malformed-raise=20guard,=20struct?= =?UTF-8?q?=20items?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../ISubscribable.cs | 4 +- .../Drivers/DriverInstanceActor.cs | 15 ++- .../DriverInstanceActorBatchForwardTests.cs | 106 ++++++++++++++++++ 3 files changed, 121 insertions(+), 4 deletions(-) diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/ISubscribable.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/ISubscribable.cs index 8e86a671..a1dd8bae 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/ISubscribable.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/ISubscribable.cs @@ -60,7 +60,9 @@ public sealed record DataChangeItem(string FullReference, DataValueSnapshot Snap /// Event payload for — every attribute /// that changed within ONE production burst (a poll cycle, an inbound message, a stream chunk). /// The handle returned by the original call. -/// The changed attributes, one entry per changed tag. Never empty. +/// The changed attributes, one entry per changed tag. Never empty. Ownership transfers +/// to the consumer, which may retain the list past the raise (it is marshalled onto another thread) — so the +/// driver MUST NOT mutate or reuse the list instance after raising; allocate a fresh one per burst. public sealed record DataChangeBatchEventArgs( ISubscriptionHandle SubscriptionHandle, IReadOnlyList Items); diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverInstanceActor.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverInstanceActor.cs index 4bb49819..98040273 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverInstanceActor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverInstanceActor.cs @@ -109,8 +109,11 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers /// (#519) — not a severity bucket. Classify it with and friends rather /// than comparing it to . public sealed record AttributeValuePublished(string DriverInstanceId, string FullReference, object? Value, uint StatusCode, DateTime TimestampUtc); - /// One value within an . - public sealed record AttributeValueItem(string FullReference, object? Value, uint StatusCode, DateTime TimestampUtc); + /// One value within an . + /// A struct deliberately: an AttributeValueItem[] then holds its items inline, so a burst of + /// N values costs one array allocation instead of one array plus N boxes. Indexed access through + /// does not box. + public readonly record struct AttributeValueItem(string FullReference, object? Value, uint StatusCode, DateTime TimestampUtc); /// Batched form of — every changed attribute of ONE /// driver production burst in one message. The parent fans the whole batch to the mux and the OPC UA /// publish path in one hop each. Status codes ride verbatim (#519). @@ -900,7 +903,13 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers // Tell, one parent hop, one mux pass per burst instead of per tag. if (subscribable is IBatchSubscribable batchable) { - _batchChangeHandler = (_, args) => self.Tell(new DataChangeBatchForward(args.Items)); + _batchChangeHandler = (_, args) => + { + // A null/empty raise from a buggy driver must never enter the mailbox: an unguarded null + // would throw in the handler, restart the actor, and tear down the driver transport. + if (args?.Items is not { Count: > 0 }) return; + self.Tell(new DataChangeBatchForward(args.Items)); + }; batchable.OnDataChangeBatch += _batchChangeHandler; } else diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverInstanceActorBatchForwardTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverInstanceActorBatchForwardTests.cs index 71a45b29..922a12fc 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverInstanceActorBatchForwardTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverInstanceActorBatchForwardTests.cs @@ -97,6 +97,10 @@ public sealed class DriverInstanceActorBatchForwardTests : RuntimeActorTestBase new DriverInstanceActor.Subscribe(new[] { "tag-a", "tag-b" }, TimeSpan.FromMilliseconds(100)), TimeSpan.FromSeconds(3)); + // Deterministic half of the assertion: the per-tag handler was never attached at all. ExpectNoMsg + // below only proves nothing arrived within its window; this proves it structurally cannot. + driver.OnDataChangeSubscriberCount.ShouldBe(0, "a batch-capable driver must NOT get the per-tag handler"); + var ts = new DateTime(2026, 8, 15, 11, 0, 0, DateTimeKind.Utc); driver.FireBatch(includePerTagRaises: true, new DataChangeItem("tag-a", new DataValueSnapshot(1, OpcUaStatus.Good, ts, ts)), @@ -172,6 +176,104 @@ public sealed class DriverInstanceActorBatchForwardTests : RuntimeActorTestBase parent.ExpectNoMsg(TimeSpan.FromMilliseconds(400)); } + /// + /// Comms loss (, Connected → Reconnecting) detaches + /// the batch handler too. The object outlives the actor across a host respawn, so a + /// handler surviving the disconnect would push a burst produced by the dead transport into a stale actor. + /// + [Fact] + public async Task DisconnectObserved_detaches_batch_handler_so_late_bursts_are_dropped() + { + var driver = new BatchSubscribableStubDriver(); + var parent = CreateTestProbe(); + parent.IgnoreMessages(m => m is DriverInstanceActor.ConnectivityChanged); + var actor = parent.ChildActorOf(DriverInstanceActor.Props(driver, reconnectInterval: TimeSpan.FromSeconds(30))); + + actor.Tell(new DriverInstanceActor.InitializeRequested("{}")); + AwaitCondition(() => driver.InitializeCount > 0, PresenceBudget); + + await actor.Ask( + new DriverInstanceActor.Subscribe(new[] { "tag-a" }, TimeSpan.FromMilliseconds(100)), + TimeSpan.FromSeconds(3)); + driver.OnDataChangeBatchSubscriberCount.ShouldBe(1); + + actor.Tell(new DriverInstanceActor.DisconnectObserved("backend went away")); + + // Race window — once the disconnect is processed the handler is gone and a late burst is inert. + AwaitCondition(() => driver.OnDataChangeBatchSubscriberCount == 0, PresenceBudget); + + var ts = new DateTime(2026, 8, 15, 13, 0, 0, DateTimeKind.Utc); + driver.FireBatch(includePerTagRaises: true, + new DataChangeItem("tag-a", new DataValueSnapshot(42, OpcUaStatus.Good, ts, ts))); + + parent.ExpectNoMsg(TimeSpan.FromMilliseconds(300)); + } + + /// + /// Stopping the actor detaches the batch handler in PostStop — the same leak guard as the + /// per-tag path, and the one that matters when the host respawns a child around the same + /// instance. + /// + [Fact] + public async Task PostStop_detaches_batch_handler() + { + var driver = new BatchSubscribableStubDriver(); + var parent = CreateTestProbe(); + parent.IgnoreMessages(m => m is DriverInstanceActor.ConnectivityChanged); + var actor = parent.ChildActorOf(DriverInstanceActor.Props(driver)); + + actor.Tell(new DriverInstanceActor.InitializeRequested("{}")); + AwaitCondition(() => driver.InitializeCount > 0, PresenceBudget); + + await actor.Ask( + new DriverInstanceActor.Subscribe(new[] { "tag-a" }, TimeSpan.FromMilliseconds(100)), + TimeSpan.FromSeconds(3)); + driver.OnDataChangeBatchSubscriberCount.ShouldBe(1); + + Sys.Stop(actor); + + AwaitCondition(() => driver.OnDataChangeBatchSubscriberCount == 0, PresenceBudget); + } + + /// + /// A malformed raise from a buggy driver — a null payload, or one carrying no items — is dropped at + /// the handler and never reaches the mailbox. Unguarded, the null would throw inside the event + /// handler, restart the actor, and take the driver transport down with it; the still-attached handler + /// (count unchanged) and the burst that lands afterwards prove neither happened. + /// + [Fact] + public async Task Malformed_batch_raise_is_dropped_without_restarting_the_actor() + { + var driver = new BatchSubscribableStubDriver(); + var parent = CreateTestProbe(); + parent.IgnoreMessages(m => m is DriverInstanceActor.ConnectivityChanged); + var actor = parent.ChildActorOf(DriverInstanceActor.Props(driver)); + + actor.Tell(new DriverInstanceActor.InitializeRequested("{}")); + AwaitCondition(() => driver.InitializeCount > 0, PresenceBudget); + + await actor.Ask( + new DriverInstanceActor.Subscribe(new[] { "tag-a" }, TimeSpan.FromMilliseconds(100)), + TimeSpan.FromSeconds(3)); + + driver.FireBatch(includePerTagRaises: false /* no items — an empty burst */); + driver.FireNullBatch(); + + parent.ExpectNoMsg(TimeSpan.FromMilliseconds(300)); + driver.OnDataChangeBatchSubscriberCount.ShouldBe(1, + "a restart would have detached the handler in PostStop"); + + // …and the subscription is still live: a well-formed burst still lands. + var ts = new DateTime(2026, 8, 15, 14, 0, 0, DateTimeKind.Utc); + driver.FireBatch(includePerTagRaises: false, + new DataChangeItem("tag-a", new DataValueSnapshot(5, OpcUaStatus.Good, ts, ts))); + + var batch = parent.ExpectMsg(TimeSpan.FromSeconds(2)); + batch.Items.Count.ShouldBe(1); + batch.Items[0].FullReference.ShouldBe("tag-a"); + batch.Items[0].Value.ShouldBe(5); + } + /// /// A that implements — it raises /// once per burst AND (optionally, per the documented driver @@ -222,5 +324,9 @@ public sealed class DriverInstanceActorBatchForwardTests : RuntimeActorTestBase OnDataChange?.Invoke(this, new DataChangeEventArgs(_handle, item.FullReference, item.Snapshot)); } } + + /// Raises the batch event with a null payload — what a buggy driver does, and what the + /// handler's guard exists to absorb. + public void FireNullBatch() => OnDataChangeBatch?.Invoke(this, null!); } }