test(core): pin batch-after-per-tag ordering; relocate batch tests beside the engine suite

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<DataChangeItem>()).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<DataChangeItem>(state.TagReferences.Count)).Add(new DataChangeItem(tagRef, current));
             }
         }
 

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;
 
 /// <summary>Contract tests for the batched data-change payload types.</summary>
 [Trait("Category", "Unit")]

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;
 
 /// <summary>
 ///     Guards for <c>PollGroupEngine</c>'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<string, object?> _values = new(StringComparer.Ordinal);
+        private readonly Dictionary<string, uint> _statusCodes = new(StringComparer.Ordinal);
         private int _pollCount;
 
         public FakeReader(params (string Reference, object? Value)[] seed)
@@ -49,6 +50,15 @@ public sealed class PollGroupEngineBatchTests
             }
         }
 
+        /// <summary>
+        ///     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).
+        /// </summary>
+        public void SetStatus(string reference, uint statusCode)
+        {
+            lock (_gate) _statusCodes[reference] = statusCode;
+        }
+
         /// <summary>The reader delegate handed to the engine.</summary>
         public Task<IReadOnlyList<DataValueSnapshot>> ReadAsync(IReadOnlyList<string> 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<IReadOnlyList<DataValueSnapshot>>(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"]);
     }
+
+    /// <summary>
+    ///     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.
+    /// </summary>
+    /// <remarks>
+    ///     <para>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.</para>
+    ///
+    ///     <para>The thread assertion covers the other way the guarantee can be lost: a batch handed
+    ///     to a detached <c>Task.Run</c> 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 <c>PollOnceAsync</c> with no <c>await</c> 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.</para>
+    /// </remarks>
+    [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<string>();
+        var perTagThreads = new ConcurrentQueue<int>();
+        var batchThreads = new ConcurrentQueue<int>();
+
+        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());
+    }
+
+    /// <summary>
+    ///     A non-zero status code is a change in its own right (the value need not move), and the
+    ///     snapshot rides into <see cref="DataChangeItem.Snapshot"/> untouched — the batch path
+    ///     never re-derives or normalises quality (#519 at its origin).
+    /// </summary>
+    [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<IReadOnlyList<DataChangeItem>>();
+
+        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);
+    }
 }