using Opc.Ua; using Opc.Ua.Server; using Shouldly; using Xunit; using ZB.MOM.WW.OtOpcUa.Core.Abstractions; using HistorianRead = ZB.MOM.WW.OtOpcUa.Core.Abstractions.HistoryReadResult; using SdkHistoryReadResult = Opc.Ua.HistoryReadResult; namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests; /// /// R08:1.1 (#736): the paged-vs-unpaged fuzz for the server raw cap. Every paging change in this repo has had boundary /// bugs that the hand-written cases missed (silent truncation, an endless cursor, an empty page between every value), so /// the cap is proved the only way that has worked: random series (bursts, gaps, sub-millisecond stamps, tie clusters), /// random windows (empty, on a tie, one step past a tie, between samples, forward / reverse / end-only, with and without /// bounds) and random caps — NumValuesPerNode 0 / 1 / 2 / small / huge, the server cap 3 … 1000 or off, the /// tie-cluster budget 3 / 8 / 65536 — paged through a REAL SDK node manager against a wire-faithful fake of each backend /// (SqlServer: tick precision, native reverse, TOP; QuestDB: µs truncation of bounds AND stamps, native reverse, /// LIMIT; Gateway: forward only, [start, end) after its own boundary-row filtering, walked back by the /// server) that enforces the backend half of the cap exactly as the real sources do (a read asking for 0 or more than /// the cap reads cap + 1 rows and is refused when the window holds more). /// /// The oracle is the same node manager's one-shot answer with the cap OFF (NumValuesPerNode = 0, no limit), /// which for a plain read is also checked against the series itself. The concatenation of the pages must equal it /// exactly; the read must terminate within the number of values plus two pages; no page may exceed the cap; the fake /// must never see a request for 0 or more than the cap (the "bounded by the same cap" invariant); and a /// BadTooManyOperations is accepted only when a tie cluster (or, on the Gateway shape, one millisecond) is /// larger than the budget that the cap and MaxTieClusterOverfetch leave — after an exact prefix. The run is /// non-vacuous by construction: it asserts that empty windows, mid-tie windows, reverse reads, NumValuesPerNode /// of 1 and 2, the smallest server cap (3) and every shape occurred, and that most iterations were exact. /// /// /// Iterations: OTOPCUA_PAGING_FUZZ_ITERATIONS (default 10 000, the lane gate). Seeds are stable, so a failure /// names a reproducible seed. /// /// public sealed class NodeManagerHistoryReadRawCapFuzzTests(BoundsServerFixture fx) : IClassFixture { private static readonly DateTime T0 = new(2026, 10, 22, 12, 0, 0, DateTimeKind.Utc); /// The backend the fake is shaped like. public enum Shape { SqlServer, QuestDb, Gateway } private enum Direction { Forward, Reverse, EndOnly } private static int Iterations => int.TryParse(Environment.GetEnvironmentVariable("OTOPCUA_PAGING_FUZZ_ITERATIONS"), out var n) && n > 0 ? n : 10_000; [Fact] public async Task Paged_equals_unpaged_on_random_series_windows_and_caps_and_always_terminates() { var iterations = Iterations; var counts = new Dictionary(); void Count(string key) => counts[key] = counts.GetValueOrDefault(key) + 1; await fx.Gate.WaitAsync(TestContext.Current.CancellationToken); try { for (var seed = 0; seed < iterations; seed++) { var rng = new Random(seed); var shape = (Shape)(seed % 3); var resolution = shape == Shape.QuestDb ? 10L : 1L; var series = RandomSeries(rng, resolution); var cap = rng.Next(10) switch { 0 => 0, 1 or 2 => 3, 3 => 4, 4 => 5, 5 => 7, 6 => 10, 7 => 16, 8 => 33, _ => 1000 }; var n = rng.Next(9) switch { 0 or 1 => 0u, 2 or 3 => 1u, 4 or 5 => 2u, 6 => (uint)rng.Next(3, 20), 7 => (uint)rng.Next(20, 200), _ => 100_000u }; var tieBudget = rng.Next(4) switch { 0 => 3, 1 => 8, _ => 65536 }; var direction = rng.Next(5) switch { 0 => Direction.Reverse, 1 => Direction.EndOnly, _ => Direction.Forward }; if (direction == Direction.EndOnly && n == 0) direction = Direction.Forward; // Part 11: one bound only with a cap var bounds = direction != Direction.EndOnly && rng.Next(4) == 0; var a = PickBoundary(rng, series, resolution); var b = PickBoundary(rng, series, resolution); var (lo, hi) = a <= b ? (a, b) : (b, a); var (start, end) = direction switch { Direction.Forward => (lo, hi), Direction.Reverse => (hi, lo), _ => (DateTime.MinValue, hi), }; var tag = $"seed {seed} {shape} cap {cap} N {n} tie {tieBudget} {direction} bounds {bounds} [{start:O} .. {end:O}]"; // The walk-back (Gateway shape only: reverse / end-only reads, and the bound lookups of any bounded read) refuses // a millisecond holding more samples than its budget — computed up front, over the whole series, because the // walk and the lookups read around the window. var walks = shape == Shape.Gateway && (direction != Direction.Forward || bounds); var msMax = walks ? MillisecondMax(series) : 0; // The oracle: the one-shot answer with the cap off (and, for a plain read, the series itself). var truthSource = new CappedFake(series, shape, cap: 0); fx.Nm.HistoricalSourceForTest(truthSource, tieBudget); // An end-only read is paged backward from EndTime (the node manager flips it to a reverse read down to the // beginning of time); its oracle is the explicit reverse read down to the Unix epoch — the walk-back's floor, // and Part 11 (the SDK checks it) needs both bounds set when NumValuesPerNode = 0. var (truth, truthError, truthCp) = direction == Direction.EndOnly ? Read(hi, DateTime.UnixEpoch, 0, bounds: false, cp: null) : Read(start, end, 0, bounds, cp: null); if (truthError.StatusCode.Code == StatusCodes.BadTooManyOperations) { // Only the walk-back can refuse an uncapped one-shot read, and only past its budget (max(1, M) with no cap). (walks && msMax > Math.Max(1, tieBudget)).ShouldBeTrue($"the oracle read was refused: {tag} (ms max {msMax})"); Count("oracle refused"); continue; } truthError.StatusCode.Code.ShouldBe(StatusCodes.Good, $"oracle: {tag}"); truthCp.ShouldBeNull($"oracle (no cap, N = 0) is one page: {tag}"); if (!bounds) Render(truth).ShouldBe(Render(Oracle(series, start, end, direction, resolution)), $"oracle vs series: {tag}"); truthSource.BadRequests.ShouldBe(0, $"a reverse window reached a forward-only backend on the oracle read: {tag}"); var windowTies = LargestTieCluster(truth); if (truth.Count == 0) Count("empty"); if (windowTies >= 2) Count("ties in window"); if (EndsOnTie(series, start, end, direction, resolution)) Count("window ends on a tie"); if (direction == Direction.Reverse) Count("reverse"); if (direction == Direction.EndOnly) Count("end-only"); if (n == 1) Count("N=1"); if (n == 2) Count("N=2"); if (n == 0) Count("N=0"); if (cap == 3) Count("cap=3"); if (bounds) Count("bounds"); Count(shape.ToString()); // A boundary tie cluster as large as the cap leaves a resumed page no room: the server steps past it with one // empty page carrying a continuation point (the documented #694 ceiling behaviour) — the one empty page allowed. var oversizedClusters = cap > 0 ? truth.Where(v => !(v.Value is null && v.StatusCode.Code == StatusCodes.BadBoundNotFound)) .GroupBy(v => v.SourceTimestamp.Ticks).Count(g => g.Count() >= Math.Max(2, cap - 1)) : 0; // The read under test. var source = new CappedFake(series, shape, cap); fx.Nm.HistoricalSourceForTest(source, tieBudget); var pages = new List>(); var collected = new List(); byte[]? cp = null; StatusCode status; do { var (values, error, next) = Read(start, end, n, bounds, cp); status = error.StatusCode; if (!StatusCode.IsGood(status)) break; pages.Add(values); collected.AddRange(values); cp = next; pages.Count.ShouldBeLessThanOrEqualTo(truth.Count + 2 + oversizedClusters, $"paging must terminate: {tag} pages {Sizes(pages)}"); } while (cp is not null); source.OverCapRequests.ShouldBe(0, $"the server asked the backend for 0 or more than the cap: {tag}"); source.BadRequests.ShouldBe(0, $"a reverse window reached a forward-only backend: {tag}"); if (status.Code == StatusCodes.BadTooManyOperations) { // Legitimate only past the budget the cap leaves: a tie cluster larger than min(M, cap − 1), or on the // Gateway shape a millisecond holding more than min(M, cap − 2) samples (the walk-back's budget). var overfetchBound = cap > 0 ? Math.Min(Math.Max(1, tieBudget), Math.Max(1, cap - 1)) : Math.Max(1, tieBudget); var walkBudget = cap > 0 ? Math.Max(1, Math.Min(tieBudget, cap - 2)) : Math.Max(1, tieBudget); var legit = windowTies > overfetchBound || (walks && msMax > walkBudget); legit.ShouldBeTrue($"refused with no tie cluster over {overfetchBound} (max {windowTies}; ms max {msMax} vs {walkBudget}): {tag}"); Render(collected).ShouldBe(Render(truth.Take(collected.Count)), $"the prefix before the refusal: {tag}"); Count("refused"); continue; } status.Code.ShouldBe(StatusCodes.Good, $"{tag}"); Render(collected).ShouldBe(Render(truth), $"paged ≠ unpaged: {tag} pages {Sizes(pages)}"); if (cap > 0) { // A bounded read whose data fits one page carries its start bound, the data and its end bound: the node // manager fits the end bound against the client's N but not against the server cap (follow-up F2), so that // one page may hold cap + 1 values; every other page is at most the cap. var bothBoundsOnOnePage = bounds && pages.Count == 1; pages.ShouldAllBe(p => p.Count <= cap + (bothBoundsOnOnePage ? 1 : 0), $"a page over the cap: {tag} pages {Sizes(pages)}"); if (bothBoundsOnOnePage && pages[0].Count == cap + 1) Count("bounded single page at cap+1 (F2)"); } if (n > 0) pages.ShouldAllBe(p => p.Count <= n, $"a page over N: {tag} pages {Sizes(pages)}"); for (var i = 0; i < pages.Count - 1; i++) { if (pages[i].Count > 0) continue; // Allowed only as the step past a boundary tie cluster the cap cannot page within. var previousLast = i > 0 && pages[i - 1].Count > 0 ? pages[i - 1][^1].SourceTimestamp.Ticks : (long?)null; var clusterAtBoundary = previousLast is { } t ? truth.Count(v => v.SourceTimestamp.Ticks == t) : 0; (cap > 0 && clusterAtBoundary >= Math.Max(2, cap - 1)).ShouldBeTrue( $"an empty page with a continuation point (page {i}, boundary cluster {clusterAtBoundary}): {tag} pages {Sizes(pages)}"); Count("empty step over an oversized tie cluster"); } if (pages.Count > 1) Count("paged"); if (pages.Count > 1 && n == 0) Count("paged at N=0"); // A cost guard against a read explosion, not a budget. A native shape reads once per page plus the bound // lookups. The Gateway shape's walk-back has its own cost profile — ~30 reads a page on a dense tag // (#691/#693), far more on a sparse series under a tiny tie budget (seed 320: 82 reads with NO cap, seed 6428: // 120+) — so it is judged RELATIVE to the same request with the cap off (the pre-#736 behaviour), per page: // the cap must not make the walk explode compared to before. The baseline asks for the page size the cap // imposes (N = cap − 1): an uncapped N = 0 reverse read used to be ONE int.MaxValue forward read — the very // memory vector the cap removes — so "what N = 0 cost before" is not the comparison. var allowedReads = 8 * pages.Count + 8; if (shape == Shape.Gateway) { var baseline = new CappedFake(series, shape, cap: 0); fx.Nm.HistoricalSourceForTest(baseline, tieBudget); var baseN = cap > 0 && (n == 0 || n > (uint)cap) ? (uint)(cap - 1) : n; var basePages = 0; byte[]? baseCp = null; do { var (_, baseError, next) = Read(start, end, baseN, bounds, baseCp); if (!StatusCode.IsGood(baseError.StatusCode)) break; basePages++; baseCp = next; } while (baseCp is not null && basePages <= truth.Count + 2 + oversizedClusters); allowedReads = 2 * baseline.Reads * Math.Max(1, pages.Count) / Math.Max(1, basePages) + 32 * pages.Count + 16; } source.Reads.ShouldBeLessThanOrEqualTo(allowedReads, $"backend reads exploded: {source.Reads} for {pages.Count} pages (allowed {allowedReads}): {tag}"); Count("exact"); } } finally { fx.Nm.HistoricalSourceForTest(NullHistorianDataSource.Instance, 65536); fx.Gate.Release(); } // Non-vacuous: every category the brief names occurred, and most runs were exact. var summary = string.Join(", ", counts.OrderBy(kv => kv.Key).Select(kv => $"{kv.Key}={kv.Value}")); foreach (var required in new[] { "empty", "ties in window", "window ends on a tie", "reverse", "end-only", "N=1", "N=2", "N=0", "cap=3", "bounds", "paged", "paged at N=0", "SqlServer", "QuestDb", "Gateway" }) counts.GetValueOrDefault(required).ShouldBeGreaterThan(iterations / 200, $"{required} under-represented: {summary}"); counts.GetValueOrDefault("exact").ShouldBeGreaterThan(iterations * 6 / 10, summary); TestContext.Current.TestOutputHelper?.WriteLine($"raw-cap fuzz: {iterations} iterations — {summary}"); } // --- generators ----------------------------------------------------------------------------- private static DateTime Ts(DataValueSnapshot s) => s.SourceTimestampUtc ?? DateTime.MinValue; private static DateTime Trunc(DateTime t, long resolution) => new(t.Ticks - t.Ticks % resolution, DateTimeKind.Utc); /// Random ascending series: bursts, gaps, sub-millisecond stamps and tie clusters; stamps stored at the resolution. private static List RandomSeries(Random rng, long resolution) { var list = new List(); var t = T0.Ticks - rng.Next(0, 3) * TimeSpan.TicksPerHour; var count = rng.Next(1, 120); var v = 0; while (list.Count < count) { t += rng.Next(6) switch { 0 => rng.Next(1, 10), // sub-millisecond (sub-µs on QuestDB: collapses onto a tie) 1 => rng.Next(1, 5) * TimeSpan.TicksPerMillisecond, // dense 2 => rng.Next(1, 3) * TimeSpan.TicksPerSecond, 3 => rng.Next(1, 30) * TimeSpan.TicksPerMinute, // a gap _ => rng.Next(1, 1000) * TimeSpan.TicksPerMillisecond, }; var ties = rng.Next(7) == 0 ? rng.Next(2, 9) : 1; for (var k = 0; k < ties; k++) { var ts = Trunc(new DateTime(t, DateTimeKind.Utc), resolution); var status = rng.Next(12) == 0 ? StatusCodes.Bad : StatusCodes.Good; list.Add(new DataValueSnapshot(status == StatusCodes.Bad ? null : (double)v, status, ts, ts)); v++; } } return list; } /// A window bound: before / after the data, exactly on a sample (a tie, maybe), one step past one, a tick before one, or between two. private static DateTime PickBoundary(Random rng, List series, long resolution) { var first = Ts(series[0]); var last = Ts(series[^1]); var i = rng.Next(series.Count); var t = Ts(series[i]); switch (rng.Next(9)) { case 0: return first.AddSeconds(-rng.Next(1, 100)); case 1: return last.AddSeconds(rng.Next(1, 100)); case 2: case 3: return t; case 4: return t.AddTicks(resolution); case 5: return t.AddTicks(-1); // sub-resolution: a microsecond backend truncates it back ONTO the sample case 6: return t.AddMilliseconds(1); default: var next = i + 1 < series.Count ? Ts(series[i + 1]) : t.AddSeconds(10); return new DateTime(t.Ticks + (long)((next.Ticks - t.Ticks) * rng.NextDouble()), DateTimeKind.Utc); } } /// The series' own answer for a plain (no bounds) read: forward [start, end), point, reverse / end-only (end, start] newest first. private static List Oracle(List series, DateTime start, DateTime end, Direction direction, long resolution) { var s = Trunc(start, resolution); var e = Trunc(end, resolution); IEnumerable window = direction switch { // start == end is Part 11's point read whichever way it was generated: the node manager reads it forward. Direction.Forward or Direction.Reverse when s == e => series.Where(x => Ts(x) == s), Direction.Forward => series.Where(x => Ts(x) >= s && Ts(x) < e), Direction.Reverse => series.Where(x => Ts(x) > e && Ts(x) <= s).Reverse(), _ => series.Where(x => Ts(x) <= e).Reverse(), }; return window.Select(x => new DataValue(new Variant(x.Value), new StatusCode(x.StatusCode), Ts(x), Ts(x))).ToList(); } /// The largest tie cluster among the returned stored values (a bound placeholder is not a stored value). private static int LargestTieCluster(IReadOnlyList truth) => truth.Where(v => !(v.Value is null && v.StatusCode.Code == StatusCodes.BadBoundNotFound)) .GroupBy(v => v.SourceTimestamp.Ticks).Select(g => g.Count()).DefaultIfEmpty(0).Max(); /// The most samples within one millisecond anywhere in the series (the walk-back's refusal rule). private static int MillisecondMax(List series) { var msMax = 0; for (int i = 0, j = 0; i < series.Count; i++) { while (Ts(series[i]).Ticks - Ts(series[j]).Ticks >= TimeSpan.TicksPerMillisecond) j++; msMax = Math.Max(msMax, i - j + 1); } return msMax; } private static bool EndsOnTie(List series, DateTime start, DateTime end, Direction direction, long resolution) { var edge = Trunc(direction == Direction.Forward ? end : direction == Direction.Reverse ? start : end, resolution); return series.Count(x => Ts(x) == edge) >= 2; } // --- helpers ---------------------------------------------------------------------------------- private static string Sizes(IEnumerable> pages) => "[" + string.Join(",", pages.Select(p => p.Count)) + "]"; private static string Render(IEnumerable values) => string.Join(" | ", values.Select(v => $"{v.SourceTimestamp.Ticks}={v.Value ?? "null"}/{v.StatusCode.Code:X8}")); private (IReadOnlyList Values, ServiceResult Error, byte[]? Cp) Read(DateTime start, DateTime end, uint max, bool bounds, byte[]? cp) { var context = new OperationContext(new RequestHeader(), secureChannelContext: null, RequestType.HistoryRead, identity: null); var details = new ReadRawModifiedDetails { StartTime = start, EndTime = end, NumValuesPerNode = max, ReturnBounds = bounds, IsReadModified = false, }; var nodesToRead = new List { new() { NodeId = fx.NodeId, ContinuationPoint = cp } }; var results = new List { null! }; var errors = new List { null! }; fx.Nm.HistoryRead(context, details, TimestampsToReturn.Both, releaseContinuationPoints: false, nodesToRead, results, errors); var values = new List(); if (results[0]?.HistoryData is { } ho && !ExtensionObject.IsNull(ho)) values.AddRange(((HistoryData)ExtensionObject.ToEncodeable(ho)).DataValues); return (values, errors[0], results[0]?.ContinuationPoint); } /// /// A series-backed historian shaped like one backend, enforcing the backend half of the server raw cap exactly as /// the real data sources do: it reports the cap as ; a read asking for 0 or more than /// the cap reads cap + 1 rows and throws when the window holds more /// (never a truncated answer). It also records every such request () — the server must /// never issue one — and every reverse window sent to the forward-only shape (). /// private sealed class CappedFake(IReadOnlyList series, Shape shape, int cap) : IHistorianDataSource { private readonly Lock _gate = new(); public int Reads { get; private set; } public int OverCapRequests { get; private set; } public int BadRequests { get; private set; } public bool SupportsReverseRaw => shape != Shape.Gateway; public long TimestampResolutionTicks => shape == Shape.QuestDb ? 10 : 1; public int MaxRawValuesPerRead => cap; private DateTime Trunc(DateTime t) => NodeManagerHistoryReadRawCapFuzzTests.Trunc(t, TimestampResolutionTicks); public Task ReadRawAsync( string fullReference, DateTime startUtc, DateTime endUtc, uint maxValuesPerNode, CancellationToken cancellationToken) { var probe = cap > 0 && (maxValuesPerNode == 0 || maxValuesPerNode > (uint)cap); lock (_gate) { Reads++; if (probe) OverCapRequests++; if (shape == Shape.Gateway && startUtc > endUtc) BadRequests++; } if (shape == Shape.Gateway && startUtc > endUtc) throw new InvalidOperationException("start_time must be less than or equal to end_time. (InvalidArgument)"); startUtc = Trunc(startUtc); endUtc = Trunc(endUtc); List window; if (startUtc > endUtc) window = series.Where(s => Ts(s) > endUtc && Ts(s) <= startUtc).Reverse().ToList(); else if (startUtc == endUtc) window = series.Where(s => Ts(s) == startUtc).ToList(); else window = series.Where(s => Ts(s) >= startUtc && Ts(s) < endUtc).ToList(); var top = probe ? cap + 1 : maxValuesPerNode == 0 ? int.MaxValue : (int)Math.Min(maxValuesPerNode, int.MaxValue); var page = window.Take(top).ToList(); if (probe && page.Count > cap) throw new HistoryReadLimitExceededException(cap, $"more than MaxRawValuesPerRead ({cap}) samples"); return Task.FromResult(new HistorianRead(page, null)); } public Task ReadProcessedAsync( string fullReference, DateTime startUtc, DateTime endUtc, TimeSpan interval, HistoryAggregateType aggregate, CancellationToken cancellationToken) => Task.FromResult(new HistorianRead(Array.Empty(), null)); public Task ReadAtTimeAsync( string fullReference, IReadOnlyList timestampsUtc, CancellationToken cancellationToken) => Task.FromResult(new HistorianRead(Array.Empty(), null)); public Task ReadEventsAsync( string? sourceName, DateTime startUtc, DateTime endUtc, int maxEvents, CancellationToken cancellationToken) => Task.FromResult(new HistoricalEventsResult(Array.Empty(), null)); public HistorianHealthSnapshot GetHealthSnapshot() => NullHistorianDataSource.Instance.GetHealthSnapshot(); public void Dispose() { } } } /// Test-only wiring shared by the paging fuzz: swap the source and the tie budget, with a fresh continuation store. internal static class NodeManagerFuzzExtensions { public static void HistoricalSourceForTest(this OtOpcUaNodeManager nm, IHistorianDataSource source, int tieBudget) { nm.HistoryContinuationStore = new InMemoryHistoryContinuationStore(); nm.MaxTieClusterOverfetch = tieBudget; nm.HistorianDataSource = source; } }