diff --git a/docs/plans/2026-08-15-value-pipeline-batching.md b/docs/plans/2026-08-15-value-pipeline-batching.md index 198e380d..8dd51a71 100644 --- a/docs/plans/2026-08-15-value-pipeline-batching.md +++ b/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 diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip/AbCipDriver.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip/AbCipDriver.cs index 78108f1d..25e2625e 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip/AbCipDriver.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip/AbCipDriver.cs @@ -24,7 +24,7 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.AbCip; /// (called in ); the library's own finalizer /// handles GC-collected tags. /// -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 { /// Authored tags this driver could not build into its poll table (Gitea #541). @@ -114,9 +114,12 @@ public sealed class AbCipDriver : IDriver, IReadable, IWritable, ITagDiscovery, /// Occurs when a subscribed tag's value changes. public event EventHandler? OnDataChange; - /// 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 once per changed tag — so - /// a consumer must subscribe to EXACTLY ONE of the two or it double-delivers. + /// + /// 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, + /// once per changed tag — so a consumer subscribes to EXACTLY + /// ONE of the two; subscribing both double-delivers. + /// public event EventHandler? OnDataChangeBatch; /// Occurs when a device's host connectivity status changes. diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.FOCAS/FocasDriver.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.FOCAS/FocasDriver.cs index 8ec0a047..67aef9eb 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.FOCAS/FocasDriver.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.FOCAS/FocasDriver.cs @@ -18,7 +18,7 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.FOCAS; /// + the default makes misconfigured servers /// fail fast. /// -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 { /// 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, /// Occurs when data changes on a subscribed tag. public event EventHandler? OnDataChange; - /// 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 once per changed tag — so - /// a consumer must subscribe to EXACTLY ONE of the two or it double-delivers. + /// + /// 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, + /// once per changed tag — so a consumer subscribes to EXACTLY + /// ONE of the two; subscribing both double-delivers. + /// public event EventHandler? OnDataChangeBatch; /// Occurs when a device host connection status changes. public event EventHandler? OnHostStatusChanged; diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus/ModbusDriver.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus/ModbusDriver.cs index 9279673a..c937c893 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus/ModbusDriver.cs +++ b/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()) .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, diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.Tests/AbCipSubscriptionTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.Tests/AbCipSubscriptionTests.cs index 901589a1..c4625681 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.Tests/AbCipSubscriptionTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.Tests/AbCipSubscriptionTests.cs @@ -20,6 +20,14 @@ public sealed class AbCipSubscriptionTests return (drv, factory); } + /// The driver advertises the batched-subscription capability (which implies ISubscribable). + [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(); + } + /// Verifies that the initial poll raises OnDataChange events for every subscribed tag. [Fact] public async Task Initial_poll_raises_OnDataChange_for_every_tag() diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.FOCAS.Tests/FocasCapabilityTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.FOCAS.Tests/FocasCapabilityTests.cs index 3bbad4e8..a6cc2623 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.FOCAS.Tests/FocasCapabilityTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.FOCAS.Tests/FocasCapabilityTests.cs @@ -47,6 +47,20 @@ public sealed class FocasCapabilityTests // ---- ISubscribable ---- + /// The driver advertises the batched-subscription capability (which implies ISubscribable). + [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(); + } + /// Verifies that the initial subscription poll raises an OnDataChange event. [Fact] public async Task Subscribe_initial_poll_raises_OnDataChange() diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus.Tests/ModbusBatchSubscriptionTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus.Tests/ModbusBatchSubscriptionTests.cs index 6d559082..9f40e07d 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus.Tests/ModbusBatchSubscriptionTests.cs +++ b/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)); } + /// + /// A per-tag subscriber that throws must not corrupt + /// the poll cycle. PollGroupEngine.PollOnceAsync invokes the driver's onChange bridge + /// UNGUARDED, so without the driver-side try/catch the throw unwinds the whole cycle before + /// onBatch 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. + /// + [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(); + // 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(ConcurrentQueue q, int target, TimeSpan timeout) { var deadline = DateTime.UtcNow + timeout;