From f81cbabd2ddcc80008926372ce9b0d05f71f3fb7 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Sat, 15 Aug 2026 12:21:42 -0400 Subject: [PATCH] test(core): pin batch-after-per-tag ordering; relocate batch tests beside the engine suite --- .../PollGroupEngine.cs | 5 +- .../DataChangeBatchContractTests.cs | 2 +- .../PollGroupEngineBatchTests.cs | 130 +++++++++++++++++- 3 files changed, 132 insertions(+), 5 deletions(-) rename tests/Core/{ZB.MOM.WW.OtOpcUa.Core.Tests => ZB.MOM.WW.OtOpcUa.Core.Abstractions.Tests}/DataChangeBatchContractTests.cs (94%) rename tests/Core/{ZB.MOM.WW.OtOpcUa.Core.Tests => ZB.MOM.WW.OtOpcUa.Core.Abstractions.Tests}/PollGroupEngineBatchTests.cs (58%) diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/PollGroupEngine.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/PollGroupEngine.cs index 0d8e3e20..8447bda4 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/PollGroupEngine.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/PollGroupEngine.cs @@ -242,7 +242,10 @@ public sealed class PollGroupEngine : IAsyncDisposable { state.LastValues[tagRef] = current; _onChange(state.Handle, tagRef, current); - if (_onBatch is not null) (changed ??= new List()).Add(new DataChangeItem(tagRef, current)); + // Capacity hint: a cycle can produce at most one item per subscribed reference, so + // sizing the list once avoids the doubling re-allocations a large group would incur. + if (_onBatch is not null) + (changed ??= new List(state.TagReferences.Count)).Add(new DataChangeItem(tagRef, current)); } } diff --git a/tests/Core/ZB.MOM.WW.OtOpcUa.Core.Tests/DataChangeBatchContractTests.cs b/tests/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions.Tests/DataChangeBatchContractTests.cs similarity index 94% rename from tests/Core/ZB.MOM.WW.OtOpcUa.Core.Tests/DataChangeBatchContractTests.cs rename to tests/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions.Tests/DataChangeBatchContractTests.cs index 738600c8..80c86802 100644 --- a/tests/Core/ZB.MOM.WW.OtOpcUa.Core.Tests/DataChangeBatchContractTests.cs +++ b/tests/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions.Tests/DataChangeBatchContractTests.cs @@ -2,7 +2,7 @@ using Shouldly; using Xunit; using ZB.MOM.WW.OtOpcUa.Core.Abstractions; -namespace ZB.MOM.WW.OtOpcUa.Core.Tests; +namespace ZB.MOM.WW.OtOpcUa.Core.Abstractions.Tests; /// Contract tests for the batched data-change payload types. [Trait("Category", "Unit")] diff --git a/tests/Core/ZB.MOM.WW.OtOpcUa.Core.Tests/PollGroupEngineBatchTests.cs b/tests/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions.Tests/PollGroupEngineBatchTests.cs similarity index 58% rename from tests/Core/ZB.MOM.WW.OtOpcUa.Core.Tests/PollGroupEngineBatchTests.cs rename to tests/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions.Tests/PollGroupEngineBatchTests.cs index a2f5acfa..4d014197 100644 --- a/tests/Core/ZB.MOM.WW.OtOpcUa.Core.Tests/PollGroupEngineBatchTests.cs +++ b/tests/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions.Tests/PollGroupEngineBatchTests.cs @@ -3,7 +3,7 @@ using Shouldly; using Xunit; using ZB.MOM.WW.OtOpcUa.Core.Abstractions; -namespace ZB.MOM.WW.OtOpcUa.Core.Tests; +namespace ZB.MOM.WW.OtOpcUa.Core.Abstractions.Tests; /// /// Guards for PollGroupEngine's optional per-cycle batch callback: one raise per poll @@ -26,6 +26,7 @@ public sealed class PollGroupEngineBatchTests { private readonly Lock _gate = new(); private readonly Dictionary _values = new(StringComparer.Ordinal); + private readonly Dictionary _statusCodes = new(StringComparer.Ordinal); private int _pollCount; public FakeReader(params (string Reference, object? Value)[] seed) @@ -49,6 +50,15 @@ public sealed class PollGroupEngineBatchTests } } + /// + /// Overwrites the status code a subsequent poll will report for a reference, leaving its + /// value untouched — the shape a device fault takes on the wire (#519). + /// + public void SetStatus(string reference, uint statusCode) + { + lock (_gate) _statusCodes[reference] = statusCode; + } + /// The reader delegate handed to the engine. public Task> ReadAsync(IReadOnlyList references, CancellationToken ct) { @@ -57,7 +67,11 @@ public sealed class PollGroupEngineBatchTests lock (_gate) { foreach (var reference in references) - snapshots.Add(new DataValueSnapshot(_values.GetValueOrDefault(reference), 0u, now, now)); + snapshots.Add(new DataValueSnapshot( + _values.GetValueOrDefault(reference), + _statusCodes.GetValueOrDefault(reference), + now, + now)); } Interlocked.Increment(ref _pollCount); return Task.FromResult>(snapshots); @@ -70,7 +84,7 @@ public sealed class PollGroupEngineBatchTests while (DateTime.UtcNow < deadline) { if (condition()) return; - await Task.Delay(10); + await Task.Delay(10, TestContext.Current.CancellationToken); } throw new TimeoutException($"Timed out after {WaitTimeout.TotalSeconds}s waiting for: {because}"); } @@ -208,4 +222,114 @@ public sealed class PollGroupEngineBatchTests batches.TryDequeue(out var changed).ShouldBeTrue(); changed!.Select(i => i.FullReference).ShouldBe(["z", "m"]); } + + /// + /// Both callbacks write into ONE ordered log, so the batch's position relative to the + /// per-tag raises of the same cycle is directly observable: in a cycle where three tags + /// moved, the batch marker must land strictly after all three per-tag markers. + /// + /// + /// This asserts ORDER, not final state — a batch raised mid-loop (after the first + /// change, say) still produces exactly the same four log entries, and is caught here only + /// because their sequence differs. A test that merely counted the entries, or compared the + /// two collections' contents, would pass against that implementation. + /// + /// The thread assertion covers the other way the guarantee can be lost: a batch handed + /// to a detached Task.Run would usually still land last in the log (nothing else is + /// appended after it), so position alone does not catch it. The engine raises both callbacks + /// in one synchronous stretch of PollOnceAsync with no await between them, so a + /// correct implementation is on the same thread by construction and this can never flake red; + /// a detached raise lands on a different pool thread. + /// + [Fact] + public async Task Batch_is_raised_after_every_per_tag_callback_of_the_same_cycle() + { + var reader = new FakeReader(("a", 1), ("b", 2), ("c", 3)); + var log = new ConcurrentQueue(); + var perTagThreads = new ConcurrentQueue(); + var batchThreads = new ConcurrentQueue(); + + await using var engine = new PollGroupEngine( + reader.ReadAsync, + // The thread id is recorded BEFORE the log marker in both callbacks: the test awaits the + // "batch" marker, so recording it second would let the assertions run against a + // not-yet-enqueued thread id. + (_, reference, _) => + { + perTagThreads.Enqueue(Environment.CurrentManagedThreadId); + log.Enqueue($"per-tag:{reference}"); + }, + minInterval: Interval, + onBatch: (_, _) => + { + batchThreads.Enqueue(Environment.CurrentManagedThreadId); + log.Enqueue("batch"); + }); + + engine.Subscribe(["a", "b", "c"], Interval); + + // Drain the initial-data cycle so the log holds exactly one cycle's markers. + await WaitForAsync(() => log.Contains("batch"), "the initial-data batch"); + log.Clear(); + perTagThreads.Clear(); + batchThreads.Clear(); + + // All three move under the reader's lock, so ONE poll cycle observes all three changes. + reader.Set(("a", 10), ("b", 20), ("c", 30)); + await WaitForAsync(() => log.Contains("batch"), "the batch for the three changed tags"); + + var ordered = log.ToArray(); + var batchIndex = Array.IndexOf(ordered, "batch"); + + // Exactly one cycle's worth: three per-tag raises then the single batch, in that order. + ordered.Count(e => e == "batch").ShouldBe(1); + ordered.Take(batchIndex).ShouldBe(["per-tag:a", "per-tag:b", "per-tag:c"]); + batchIndex.ShouldBe(3); + + // Raised inline on the polling thread, not posted to a detached task. + batchThreads.Count.ShouldBe(1); + batchThreads.Single().ShouldBe(perTagThreads.Last()); + } + + /// + /// A non-zero status code is a change in its own right (the value need not move), and the + /// snapshot rides into untouched — the batch path + /// never re-derives or normalises quality (#519 at its origin). + /// + [Fact] + public async Task Batch_carries_a_bad_status_snapshot_unchanged() + { + const uint BadStatus = 0x80010000u; // Bad_UnexpectedError + + var reader = new FakeReader(("a", 1), ("b", 2)); + var perTag = new ConcurrentQueue<(string Reference, DataValueSnapshot Snapshot)>(); + var batches = new ConcurrentQueue>(); + + await using var engine = new PollGroupEngine( + reader.ReadAsync, + (_, reference, snapshot) => perTag.Enqueue((reference, snapshot)), + minInterval: Interval, + onBatch: (_, items) => batches.Enqueue(items)); + + engine.Subscribe(["a", "b"], Interval); + + await WaitForAsync(() => batches.Count >= 1, "the initial-data batch"); + perTag.Clear(); + batches.Clear(); + + // The device faults on "a": same value, bad quality. + reader.SetStatus("a", BadStatus); + + await WaitForAsync(() => batches.Count >= 1, "the batch for the newly-bad tag"); + batches.TryDequeue(out var bad).ShouldBeTrue(); + bad!.Count.ShouldBe(1); + bad[0].FullReference.ShouldBe("a"); + bad[0].Snapshot.StatusCode.ShouldBe(BadStatus); + bad[0].Snapshot.Value.ShouldBe(1); // value unchanged — the status alone drove the raise + + // The per-tag callback saw the identical snapshot instance. + perTag.TryDequeue(out var perTagItem).ShouldBeTrue(); + perTagItem.Reference.ShouldBe("a"); + perTagItem.Snapshot.ShouldBeSameAs(bad[0].Snapshot); + } }