From cd4ce305a2f98ec5666b8efbbad37e1c09064779 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Sat, 15 Aug 2026 12:17:54 -0400 Subject: [PATCH] =?UTF-8?q?fix(modbus):=20batch=20event=20honors=20deadban?= =?UTF-8?q?d=20=E2=80=94=20raise=20from=20the=20driver-filtered=20buffer?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../ModbusDriver.cs | 32 ++++++++- .../ModbusBatchSubscriptionTests.cs | 68 +++++++++++++++++++ 2 files changed, 99 insertions(+), 1 deletion(-) 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 543a199c..9279673a 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus/ModbusDriver.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus/ModbusDriver.cs @@ -53,6 +53,17 @@ public sealed class ModbusDriver // does a typed cast inside. private readonly ConcurrentDictionary _lastPublishedByRef = new(StringComparer.OrdinalIgnoreCase); + // Per-cycle buffer of the items this driver actually PUBLISHED (i.e. that passed ShouldPublish), + // keyed by subscription handle. The poll engine collects its own batch list unconditionally — it + // has no knowledge of the deadband filter — so the batch event is raised from this buffer instead, + // using the engine's onBatch purely as the end-of-cycle signal. + // + // No lock is needed on the per-handle List: the engine runs one loop task per subscription, so for + // ONE handle all of a cycle's onChange calls and that cycle's single onBatch run sequentially on + // that subscription's own task. Different subscriptions run concurrently — hence the per-handle + // ConcurrentDictionary. + private readonly ConcurrentDictionary> _cycleBatchItems = new(); + // Last-written value per tag for the WriteOnChangeOnly suppression. Invalidated by reads // that return a different value (so an HMI-side change doesn't get masked). private readonly Dictionary _lastWrittenByRef = new(StringComparer.OrdinalIgnoreCase); @@ -129,11 +140,25 @@ public sealed class ModbusDriver onChange: (handle, tagRef, snapshot) => { if (!ShouldPublish(tagRef, snapshot)) return; + // Buffer what we publish so the cycle-end batch carries exactly the same items. + _cycleBatchItems.GetOrAdd(handle, _ => new List()) + .Add(new DataChangeItem(tagRef, snapshot)); OnDataChange?.Invoke(this, new DataChangeEventArgs(handle, tagRef, snapshot)); }, onError: HandlePollError, backoffCap: PollBackoffCap, - onBatch: (handle, items) => OnDataChangeBatch?.Invoke(this, new DataChangeBatchEventArgs(handle, items))); + // The engine's own items list is deliberately IGNORED: it is collected pre-deadband + // (PollGroupEngine has no knowledge of ShouldPublish), so forwarding it would carry + // changes the per-tag event suppressed — and a consumer that prefers the batch event + // would see deadband silently stop working. The buffer holds exactly what the per-tag + // event published, keeping the two events content-identical per the IBatchSubscribable + // contract. onBatch is used only as the end-of-cycle signal. A cycle whose every change + // was suppressed leaves the buffer empty and raises nothing. + onBatch: (handle, _) => + { + if (_cycleBatchItems.TryRemove(handle, out var published) && published.Count > 0) + OnDataChangeBatch?.Invoke(this, new DataChangeBatchEventArgs(handle, published)); + }); } /// Upper bound on the poll-loop failure backoff — adopts the S7-proven 30 s cap fleet-wide (05/STAB-8). @@ -1215,6 +1240,9 @@ public sealed class ModbusDriver public Task UnsubscribeAsync(ISubscriptionHandle handle, CancellationToken cancellationToken) { _poll.Unsubscribe(handle); + // Drop any partially-collected cycle buffer so a torn-down subscription leaks nothing. + // Unsubscribe awaits the loop task, so no onChange can re-add after this point. + _cycleBatchItems.TryRemove(handle, out _); return Task.CompletedTask; } @@ -1656,6 +1684,8 @@ public sealed class ModbusDriver lock (_autoProhibitedLock) _autoProhibited.Clear(); await _poll.DisposeAsync().ConfigureAwait(false); + // After the engine's loops are drained, no onChange can re-populate the per-cycle buffers. + _cycleBatchItems.Clear(); if (_transport is not null) await _transport.DisposeAsync().ConfigureAwait(false); _transport = null; 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 ee2535ec..6d559082 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 @@ -129,6 +129,74 @@ public sealed class ModbusBatchSubscriptionTests } } + /// + /// The batch event honours the driver's own per-tag deadband filter. The + /// PollGroupEngine collects batch items unconditionally — it knows nothing about + /// ShouldPublish — so a straight pass-through of the engine's list would carry + /// changes the per-tag event suppressed, and a downstream consumer that PREFERS the batch + /// event would see deadband silently stop working. A cycle whose only changes are all + /// suppressed raises no batch at all. + /// + [Fact] + public async Task Deadband_suppressed_changes_are_excluded_from_the_batch() + { + var (drv, fake) = NewDriver( + new ModbusTagDefinition("Temp", ModbusRegion.HoldingRegisters, 0, ModbusDataType.Int16, Deadband: 5.0)); + await drv.InitializeAsync("{}", CancellationToken.None); + fake.HoldingRegisters[0] = 100; + + var perTag = new ConcurrentQueue(); + var batches = new ConcurrentQueue(); + drv.OnDataChange += (_, e) => perTag.Enqueue(e); + ((IBatchSubscribable)drv).OnDataChangeBatch += (_, e) => batches.Enqueue(e); + + var handle = await drv.SubscribeAsync(["Temp"], TimeSpan.FromMilliseconds(100), CancellationToken.None); + try + { + // Cycle 1 — initial-data push. First sample seeds the deadband baseline and publishes + // through both events. + await WaitForCountAsync(batches, 1, TimeSpan.FromSeconds(2)); + batches.Count.ShouldBe(1); + batches.Single().Items.Count.ShouldBe(1); + batches.Single().Items[0].Snapshot.Value.ShouldBe((short)100); + perTag.Count.ShouldBe(1); + + // Under-threshold change (delta 2 < 5): the ENGINE sees a value change and would offer + // it as a batch item, but the driver's deadband suppresses the publish — so neither + // event fires and no batch is raised for that cycle at all. + fake.HoldingRegisters[0] = 102; + await Task.Delay(600); + perTag.Count.ShouldBe(1, "under-threshold change is suppressed on the per-tag event"); + batches.Count.ShouldBe(1, "a cycle whose only change is deadband-suppressed raises no batch"); + + // Over-threshold change (delta 10 ≥ 5 against the 100 baseline): publishes on both. + fake.HoldingRegisters[0] = 110; + 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.ShouldBe(2); + allBatches[1].Items.Count.ShouldBe(1); + allBatches[1].Items[0].Snapshot.Value.ShouldBe((short)110); + + // Content equality: flattening the batches reproduces the per-tag event stream exactly. + // The suppressed 102 appears in NEITHER. + var flattened = allBatches.SelectMany(b => b.Items).ToArray(); + var perTagArray = perTag.ToArray(); + flattened.Length.ShouldBe(perTagArray.Length); + for (var i = 0; i < flattened.Length; i++) + { + flattened[i].FullReference.ShouldBe(perTagArray[i].FullReference); + flattened[i].Snapshot.ShouldBe(perTagArray[i].Snapshot); + } + flattened.ShouldNotContain(item => Equals(item.Snapshot.Value, (short)102)); + } + private static async Task WaitForCountAsync(ConcurrentQueue q, int target, TimeSpan timeout) { var deadline = DateTime.UtcNow + timeout;