fix(runtime): harden batch forwarding — detach pins, malformed-raise guard, struct items

src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/ISubscribable.cs

@@ -60,7 +60,9 @@ public sealed record DataChangeItem(string FullReference, DataValueSnapshot Snap
 /// <summary>Event payload for <see cref="IBatchSubscribable.OnDataChangeBatch"/> — every attribute
 /// that changed within ONE production burst (a poll cycle, an inbound message, a stream chunk).</summary>
 /// <param name="SubscriptionHandle">The handle returned by the original <see cref="ISubscribable.SubscribeAsync"/> call.</param>
-/// <param name="Items">The changed attributes, one entry per changed tag. Never empty.</param>
+/// <param name="Items">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.</param>
 public sealed record DataChangeBatchEventArgs(
     ISubscriptionHandle SubscriptionHandle,
     IReadOnlyList<DataChangeItem> Items);

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 <see cref="OpcUaStatus.IsGood"/> and friends rather
     /// than comparing it to <see cref="OpcUaStatus.Good"/>.</para></summary>
     public sealed record AttributeValuePublished(string DriverInstanceId, string FullReference, object? Value, uint StatusCode, DateTime TimestampUtc);
-    /// <summary>One value within an <see cref="AttributeValueBatchPublished"/>.</summary>
-    public sealed record AttributeValueItem(string FullReference, object? Value, uint StatusCode, DateTime TimestampUtc);
+    /// <summary>One value within an <see cref="AttributeValueBatchPublished"/>.
+    /// <para>A struct deliberately: an <c>AttributeValueItem[]</c> 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
+    /// <see cref="IReadOnlyList{T}"/> does not box.</para></summary>
+    public readonly record struct AttributeValueItem(string FullReference, object? Value, uint StatusCode, DateTime TimestampUtc);
     /// <summary>Batched form of <see cref="AttributeValuePublished"/> — 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).</summary>
@@ -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

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));
     }
 
+    /// <summary>
+    ///     Comms loss (<see cref="DriverInstanceActor.DisconnectObserved"/>, Connected → Reconnecting) detaches
+    ///     the batch handler too. The <see cref="IDriver"/> 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.
+    /// </summary>
+    [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<DriverInstanceActor.SubscriptionEstablished>(
+            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));
+    }
+
+    /// <summary>
+    ///     Stopping the actor detaches the batch handler in <c>PostStop</c> — the same leak guard as the
+    ///     per-tag path, and the one that matters when the host respawns a child around the same
+    ///     <see cref="IDriver"/> instance.
+    /// </summary>
+    [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<DriverInstanceActor.SubscriptionEstablished>(
+            new DriverInstanceActor.Subscribe(new[] { "tag-a" }, TimeSpan.FromMilliseconds(100)),
+            TimeSpan.FromSeconds(3));
+        driver.OnDataChangeBatchSubscriberCount.ShouldBe(1);
+
+        Sys.Stop(actor);
+
+        AwaitCondition(() => driver.OnDataChangeBatchSubscriberCount == 0, PresenceBudget);
+    }
+
+    /// <summary>
+    ///     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.
+    /// </summary>
+    [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<DriverInstanceActor.SubscriptionEstablished>(
+            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<DriverInstanceActor.AttributeValueBatchPublished>(TimeSpan.FromSeconds(2));
+        batch.Items.Count.ShouldBe(1);
+        batch.Items[0].FullReference.ShouldBe("tag-a");
+        batch.Items[0].Value.ShouldBe(5);
+    }
+
     /// <summary>
     ///     A <see cref="StubDriver"/> that implements <see cref="IBatchSubscribable"/> — it raises
     ///     <see cref="OnDataChangeBatch"/> 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));
             }
         }
+
+        /// <summary>Raises the batch event with a null payload — what a buggy driver does, and what the
+        /// handler's guard exists to absorb.</summary>
+        public void FireNullBatch() => OnDataChangeBatch?.Invoke(this, null!);
     }
 }