fix(modbus): exception-safe batch wrapper; driver batch wiring consistency

docs/plans/2026-08-15-value-pipeline-batching.md

@@ -217,6 +217,20 @@ Per-tag raising is already wired via the engine's `onChange` — leave it.
 
 **Step 5:** `git add -A && git commit -m "feat(drivers): Modbus + S7 implement IBatchSubscribable via PollGroupEngine batch callback"`
 
+**Deviation (landed):** Modbus does **not** forward the engine's `items` argument — it ignores it and
+raises the batch from its own per-cycle buffer of the items that passed `ShouldPublish`. The engine
+collects its batch list unconditionally and has no knowledge of the driver's deadband filter, so a
+straight pass-through would carry changes the per-tag event suppressed, and a consumer that preferred
+the batch event would see deadband silently stop working (review finding). `onBatch` is therefore used
+only as the end-of-cycle signal, draining the buffer; a cycle whose every change was suppressed raises
+nothing. Because that buffer is the only piece of cross-cycle state any of the four poll drivers holds,
+the driver's `onChange` bridge also wraps `OnDataChange?.Invoke(...)` in a swallowing try/catch
+(mirroring `PollGroupEngine.ReportError`): the engine invokes the bridge unguarded, so a throwing
+external subscriber would unwind the cycle before `onBatch` ran and the undrained items would merge
+into a later cycle's batch. The buffer append stays outside the catch — the item was published as far
+as the batch contract is concerned; only the per-tag subscriber faulted. S7/AbCip/FOCAS forward the
+engine's list as this task describes and hold no such state.
+
 ---
 
 ### Task 3b: AbCip + FOCAS raise the batch event

src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip/AbCipDriver.cs

@@ -24,7 +24,7 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.AbCip;
 ///     (called in <see cref="DeviceState.DisposeHandles"/>); the library's own finalizer
 ///     handles GC-collected tags.</para>
 /// </remarks>
-public sealed class AbCipDriver : IDriver, IReadable, IWritable, ITagDiscovery, ISubscribable, IBatchSubscribable,
+public sealed class AbCipDriver : IDriver, IReadable, IWritable, ITagDiscovery, IBatchSubscribable,
     IHostConnectivityProbe, IPerCallHostResolver, IAlarmSource, ISkippedTagReporter, IDisposable, IAsyncDisposable
 {
     /// <summary>Authored tags this driver could not build into its poll table (Gitea #541).</summary>
@@ -114,9 +114,12 @@ public sealed class AbCipDriver : IDriver, IReadable, IWritable, ITagDiscovery,
     /// <summary>Occurs when a subscribed tag's value changes.</summary>
     public event EventHandler<DataChangeEventArgs>? OnDataChange;
 
-    /// <summary>Occurs once per poll cycle carrying every tag that changed in that cycle. The driver raises
-    /// BOTH events for every burst — this one once, and <see cref="OnDataChange"/> once per changed tag — so
-    /// a consumer must subscribe to EXACTLY ONE of the two or it double-delivers.</summary>
+    /// <summary>
+    ///     Occurs once per poll cycle that produced at least one change, carrying that cycle's
+    ///     changed tags. Both data-change events fire by design — this one once per cycle,
+    ///     <see cref="OnDataChange"/> once per changed tag — so a consumer subscribes to EXACTLY
+    ///     ONE of the two; subscribing both double-delivers.
+    /// </summary>
     public event EventHandler<DataChangeBatchEventArgs>? OnDataChangeBatch;
 
     /// <summary>Occurs when a device's host connectivity status changes.</summary>

src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.FOCAS/FocasDriver.cs

@@ -18,7 +18,7 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.FOCAS;
 ///     + the default <see cref="UnimplementedFocasClientFactory"/> makes misconfigured servers
 ///     fail fast.
 /// </remarks>
-public sealed class FocasDriver : IDriver, IReadable, IWritable, ITagDiscovery, ISubscribable, IBatchSubscribable,
+public sealed class FocasDriver : IDriver, IReadable, IWritable, ITagDiscovery, IBatchSubscribable,
     IHostConnectivityProbe, IPerCallHostResolver, IAlarmSource, ISkippedTagReporter, IDisposable, IAsyncDisposable
 {
     /// <summary>Authored tags this driver could not build into its poll table (Gitea #541). FOCAS has three
@@ -80,9 +80,12 @@ public sealed class FocasDriver : IDriver, IReadable, IWritable, ITagDiscovery,
 
     /// <summary>Occurs when data changes on a subscribed tag.</summary>
     public event EventHandler<DataChangeEventArgs>? OnDataChange;
-    /// <summary>Occurs once per poll cycle carrying every tag that changed in that cycle. The driver raises
-    /// BOTH events for every burst — this one once, and <see cref="OnDataChange"/> once per changed tag — so
-    /// a consumer must subscribe to EXACTLY ONE of the two or it double-delivers.</summary>
+    /// <summary>
+    ///     Occurs once per poll cycle that produced at least one change, carrying that cycle's
+    ///     changed tags. Both data-change events fire by design — this one once per cycle,
+    ///     <see cref="OnDataChange"/> once per changed tag — so a consumer subscribes to EXACTLY
+    ///     ONE of the two; subscribing both double-delivers.
+    /// </summary>
     public event EventHandler<DataChangeBatchEventArgs>? OnDataChangeBatch;
     /// <summary>Occurs when a device host connection status changes.</summary>
     public event EventHandler<HostStatusChangedEventArgs>? OnHostStatusChanged;

src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus/ModbusDriver.cs

@@ -141,9 +141,19 @@ public sealed class ModbusDriver
             {
                 if (!ShouldPublish(tagRef, snapshot)) return;
                 // Buffer what we publish so the cycle-end batch carries exactly the same items.
+                // The append stays OUTSIDE the catch below on purpose: as far as the batch contract
+                // is concerned the item WAS published — only the per-tag subscriber faulted — so it
+                // must still appear in this cycle's batch, keeping the two events content-identical.
                 _cycleBatchItems.GetOrAdd(handle, _ => new List<DataChangeItem>())
                     .Add(new DataChangeItem(tagRef, snapshot));
-                OnDataChange?.Invoke(this, new DataChangeEventArgs(handle, tagRef, snapshot));
+                // Defensive, mirroring PollGroupEngine.ReportError: the engine invokes this bridge
+                // UNGUARDED, so a throwing external subscriber would unwind PollOnceAsync mid-loop
+                // and the cycle would never reach onBatch. Because the buffer above persists across
+                // cycles (it is drained only by onBatch), those items would then silently MERGE into
+                // a later cycle's batch — cross-cycle contamination unique to this driver. Swallowing
+                // is what keeps the cycle running so onBatch drains the buffer on schedule.
+                try { OnDataChange?.Invoke(this, new DataChangeEventArgs(handle, tagRef, snapshot)); }
+                catch { /* a buggy subscriber must not corrupt the poll cycle */ }
             },
             onError: HandlePollError,
             backoffCap: PollBackoffCap,

tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.Tests/AbCipSubscriptionTests.cs

@@ -20,6 +20,14 @@ public sealed class AbCipSubscriptionTests
         return (drv, factory);
     }
 
+    /// <summary>The driver advertises the batched-subscription capability (which implies ISubscribable).</summary>
+    [Fact]
+    public void Driver_is_batch_subscribable()
+    {
+        var (drv, _) = NewDriver(new AbCipTagDefinition("Speed", "ab://10.0.0.5/1,0", "Speed", AbCipDataType.DInt));
+        (drv is IBatchSubscribable).ShouldBeTrue();
+    }
+
     /// <summary>Verifies that the initial poll raises OnDataChange events for every subscribed tag.</summary>
     [Fact]
     public async Task Initial_poll_raises_OnDataChange_for_every_tag()

tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.FOCAS.Tests/FocasCapabilityTests.cs

@@ -47,6 +47,20 @@ public sealed class FocasCapabilityTests
 
     // ---- ISubscribable ----
 
+    /// <summary>The driver advertises the batched-subscription capability (which implies ISubscribable).</summary>
+    [Fact]
+    public void Driver_is_batch_subscribable()
+    {
+        var drv = new FocasDriver(new FocasDriverOptions
+        {
+            Devices = [new FocasDeviceOptions("focas://10.0.0.5:8193")],
+            RawTags = [FocasRawTags.Entry(new FocasTagDefinition("X", "focas://10.0.0.5:8193", "R100", FocasDataType.Byte))],
+            Probe = new FocasProbeOptions { Enabled = false },
+        }, "drv-1", new FakeFocasClientFactory());
+
+        (drv is IBatchSubscribable).ShouldBeTrue();
+    }
+
     /// <summary>Verifies that the initial subscription poll raises an OnDataChange event.</summary>
     [Fact]
     public async Task Subscribe_initial_poll_raises_OnDataChange()

tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus.Tests/ModbusBatchSubscriptionTests.cs

@@ -197,6 +197,74 @@ public sealed class ModbusBatchSubscriptionTests
         flattened.ShouldNotContain(item => Equals(item.Snapshot.Value, (short)102));
     }
 
+    /// <summary>
+    ///     A per-tag <see cref="ISubscribable.OnDataChange"/> subscriber that throws must not corrupt
+    ///     the poll cycle. <c>PollGroupEngine.PollOnceAsync</c> invokes the driver's onChange bridge
+    ///     UNGUARDED, so without the driver-side try/catch the throw unwinds the whole cycle before
+    ///     <c>onBatch</c> runs — and because Modbus buffers its published items across the cycle
+    ///     (the deadband-honouring buffer the tests above pin), the undrained items would silently
+    ///     MERGE into a later cycle's batch. This failure mode is unique to Modbus: the other poll
+    ///     drivers forward the engine's own per-cycle list and hold no state between cycles.
+    ///     Asserted on both the shape (one batch per cycle, three items each) and the VALUES, so a
+    ///     merged or dropped batch fails rather than coincidentally matching on count.
+    /// </summary>
+    [Fact]
+    public async Task A_throwing_per_tag_subscriber_does_not_break_or_contaminate_the_cycle_batch()
+    {
+        var (drv, fake) = NewDriver(
+            new ModbusTagDefinition("Level", ModbusRegion.HoldingRegisters, 0, ModbusDataType.Int16),
+            new ModbusTagDefinition("Temp", ModbusRegion.HoldingRegisters, 1, ModbusDataType.Int16),
+            new ModbusTagDefinition("Flow", ModbusRegion.HoldingRegisters, 2, ModbusDataType.Int16));
+        await drv.InitializeAsync("{}", CancellationToken.None);
+        fake.HoldingRegisters[0] = 100;
+        fake.HoldingRegisters[1] = 200;
+        fake.HoldingRegisters[2] = 300;
+
+        var thrown = 0;
+        var batches = new ConcurrentQueue<DataChangeBatchEventArgs>();
+        // The buggy external subscriber: every per-tag delivery faults.
+        drv.OnDataChange += (_, _) =>
+        {
+            Interlocked.Increment(ref thrown);
+            throw new InvalidOperationException("buggy per-tag subscriber");
+        };
+        ((IBatchSubscribable)drv).OnDataChangeBatch += (_, e) => batches.Enqueue(e);
+
+        var handle = await drv.SubscribeAsync(["Level", "Temp", "Flow"], TimeSpan.FromMilliseconds(100), CancellationToken.None);
+        try
+        {
+            // Cycle 1 — the initial-data push. All three tags publish despite every per-tag
+            // delivery faulting, and the cycle still reaches its batch.
+            await WaitForCountAsync(batches, 1, TimeSpan.FromSeconds(2));
+            batches.Count.ShouldBe(1, "the cycle must still reach onBatch after a subscriber throws");
+
+            // Cycle 2 — every tag changes again.
+            fake.HoldingRegisters[0] = 101;
+            fake.HoldingRegisters[1] = 201;
+            fake.HoldingRegisters[2] = 301;
+            await WaitForCountAsync(batches, 2, TimeSpan.FromSeconds(3));
+        }
+        finally
+        {
+            // Unsubscribe drains the in-flight poll loop, so the queues are stable from here.
+            await drv.UnsubscribeAsync(handle, CancellationToken.None);
+        }
+
+        var allBatches = batches.ToArray();
+        allBatches.Length.ShouldBeGreaterThanOrEqualTo(2);
+
+        allBatches[0].Items.Count.ShouldBe(3);
+        allBatches[0].Items.Select(i => i.Snapshot.Value).ShouldBe(new object[] { (short)100, (short)200, (short)300 });
+
+        // The load-bearing assertion: cycle 2's batch carries EXACTLY cycle 2's items. A cycle
+        // aborted mid-loop would have left its items in the buffer to merge in here.
+        allBatches[1].Items.Count.ShouldBe(3, "a cycle aborted before onBatch would merge its buffered items into the next batch");
+        allBatches[1].Items.Select(i => i.Snapshot.Value).ShouldBe(new object[] { (short)101, (short)201, (short)301 });
+
+        // Sanity: the throwing subscriber really was invoked for every published item.
+        thrown.ShouldBeGreaterThanOrEqualTo(6);
+    }
+
     private static async Task WaitForCountAsync<T>(ConcurrentQueue<T> q, int target, TimeSpan timeout)
     {
         var deadline = DateTime.UtcNow + timeout;