dohertj2/lmxopcua
1
0
Fork 0
Code Issues 14 Pull Requests Actions Packages Projects Releases Wiki Activity

fix(historian): #689 size reverse-walk segments from the page, not the tie budget

Browse Source 
ReverseRawReader aimed every segment at half of MaxTieClusterOverfetch
(~32k rows by default) and carried that width page to page, so a paged
'latest N' read on a dense Gateway tag fetched ~32,770 rows per page
(2-5 s live) to return N. The walk now lives in a generic ReverseWalk
core: segments are aimed at ~N+8 rows (page budget 2N+16), a discovery
segment (first without a hint / after an empty one) may read 1,024, and
the tie budget is only the ceiling a single-timestamp cluster escalates
to. A truncated read never re-enters the stretch below its newest row
until the gap above it is bisected. Tests: per-page row counts on a
dense tag (node manager + walk), randomized exactness fuzz (walk and
node-manager paging), dense-tag-below-a-long-gap read bound.
This commit is contained in:
Joseph Doherty
2026-10-12 16:25:07 -04:00
parent 71d031fcd0
commit d33992034d
8 changed files with 663 additions and 183 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadPagingTests.cs
+102 -3
View File
@@ -650,6 +650,95 @@ public sealed class NodeManagerHistoryReadPagingTests : IDisposable
await host.DisposeAsync();
}
/// <summary>
/// #689: a paged end-only read on a dense tag sizes each segment from the PAGE (N), not from
/// <c>MaxTieClusterOverfetch</c>. Live (AVEVA, 1 Hz, N = 3): from about page 5 every page read 32,770 rows and took
/// 2–5 s to return 2 rows, because the walk re-aimed its width at half the 65,536-row tie budget and carried it.
/// </summary>
[Fact]
public async Task Raw_reverse_paging_on_a_dense_forward_only_tag_reads_about_the_page_size_per_page()
{
// 100,000 samples 100 ms apart (~2.8 h): far more than the default tie budget would ever let a page read.
var series = Enumerable.Range(0, 100_000)
.Select(i => { var t = Epoch.AddMilliseconds(100L * i); return new DataValueSnapshot((double)i, StatusCodes.Good, t, t); })
.ToArray();
var source = new ForwardOnlyHistorianDataSource(series);
var (host, nm, nodeId) = await BootForwardOnlyAsync(source, "FwdDense689");
nm.MaxTieClusterOverfetch.ShouldBe(65536, "the default budget — the one the live gate ran with");
var end = Epoch.AddMilliseconds(100L * 99_999);
byte[]? cp = null;
var expected = 99_999;
for (var page = 1; page <= 20; page++)
{
source.ResetCounts();
var (values, error, next) = ReadRaw(nm, nodeId, DateTime.MinValue, end, max: 3, inboundCp: cp);
error.StatusCode.Code.ShouldBe(StatusCodes.Good);
values.ShouldNotBeEmpty();
values[^1].ShouldBe((double)(expected - 2), $"page {page} continues exactly");
expected = (int)values[^1];
next.ShouldNotBeNull();
cp = next;
// The discriminating counts: rows read and the largest cap asked for, per page. Page 1 has no density yet and
// may read one discovery segment (a minute here, 600 rows); every page after it is sized from N.
var rowsBound = page == 1 ? ReverseWalk.DiscoveryRows + 2 : 64;
source.RowsReturned.ShouldBeLessThanOrEqualTo(rowsBound, $"page {page} read {source.RowsReturned} rows to return {values.Count}");
source.MaxCapRequested.ShouldBeLessThanOrEqualTo((uint)rowsBound, $"page {page} asked for {source.MaxCapRequested} rows");
source.ReadCount.ShouldBeLessThanOrEqualTo(3, $"page {page}");
}
await host.DisposeAsync();
}
/// <summary>
/// #689/#691: randomized end-to-end paging of end-only reads through the node manager on a forward-only backend —
/// bursts, gaps, sub-millisecond stamps and tie clusters (some larger than the page) — the union of the pages is
/// every sample exactly once, newest first, whatever N and the tie budget.
/// </summary>
[Fact]
public async Task Raw_reverse_paging_on_a_forward_only_backend_is_exact_on_random_data()
{
var (host, nm, nodeId) = await BootForwardOnlyAsync(new ForwardOnlyHistorianDataSource([]), "FwdFuzz");
for (var seed = 0; seed < 150; seed++)
{
var rng = new Random(seed);
var series = new List<DataValueSnapshot>();
var t = Epoch.Ticks;
var v = 0;
var count = rng.Next(1, 300);
while (series.Count < count)
{
t += rng.Next(5) switch
{
0 => rng.Next(1, 10),
1 => rng.Next(1, 5) * TimeSpan.TicksPerMillisecond,
2 => rng.Next(1, 90) * TimeSpan.TicksPerMinute,
_ => rng.Next(1, 2000) * TimeSpan.TicksPerMillisecond,
};
var ties = rng.Next(6) == 0 ? rng.Next(2, 7) : 1;
for (var k = 0; k < ties; k++)
{
var ts = new DateTime(t, DateTimeKind.Utc);
series.Add(new DataValueSnapshot((double)v++, StatusCodes.Good, ts, ts));
}
}
// No fabricated start row: the Gateway source drops it (#687) — the node manager's tie-cluster point read relies on
// that. The walk itself is fuzzed WITH the row in ReverseWalkTests.
nm.HistorianDataSource = new ForwardOnlyHistorianDataSource(series, emitBoundaryRow: false);
nm.MaxTieClusterOverfetch = rng.Next(2) == 0 ? 8 : 65536;
var max = (uint)rng.Next(1, 12);
var end = new DateTime(t + rng.Next(0, 3) * TimeSpan.TicksPerSecond, DateTimeKind.Utc);
var collected = await DrainAsync(nm, nodeId, DateTime.MinValue, end, max, maxPages: 2000);
collected.ShouldBe(Enumerable.Range(0, series.Count).Reverse().Select(i => (double)i), $"seed {seed}, N = {max}");
}
await host.DisposeAsync();
}
/// <summary>#686 review: a reverse read from <c>StartTime = DateTime.MaxValue</c> walks back from now + 1 day, not from
/// year 9999.</summary>
[Fact]
@@ -978,9 +1067,15 @@ public sealed class NodeManagerHistoryReadPagingTests : IDisposable
/// <summary>Reads whose row count equalled the cap (a full, possibly truncated, read).</summary>
public int FullReads { get; private set; }
/// <summary>Rows returned across every read since the last <see cref="ResetCounts"/> (#689).</summary>
public long RowsReturned { get; private set; }
/// <summary>The largest <c>maxValuesPerNode</c> asked for since the last <see cref="ResetCounts"/> (#689).</summary>
public uint MaxCapRequested { get; private set; }
public void ResetCounts()
{
lock (_gate) FullReads = 0;
lock (_gate) { FullReads = 0; RowsReturned = 0; MaxCapRequested = 0; }
Interlocked.Exchange(ref _readCount, 0);
}
@@ -1016,8 +1111,12 @@ public sealed class NodeManagerHistoryReadPagingTests : IDisposable
: new DataValueSnapshot(prior.Value, prior.StatusCode, stamp, stamp));
rows.AddRange(series.Where(s => Ts(s) >= startUtc && Ts(s) <= endUtc));
var page = maxValuesPerNode == 0 ? rows : rows.Take((int)maxValuesPerNode).ToList();
if (maxValuesPerNode > 0 && page.Count == maxValuesPerNode)
lock (_gate) FullReads++;
lock (_gate)
{
if (maxValuesPerNode > 0 && page.Count == maxValuesPerNode) FullReads++;
RowsReturned += page.Count;
if (maxValuesPerNode > MaxCapRequested) MaxCapRequested = maxValuesPerNode;
}
int seen;
while (page.Count > (seen = Volatile.Read(ref _maxRowsReturned))
&& Interlocked.CompareExchange(ref _maxRowsReturned, page.Count, seen) != seen) { }
tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/ReverseWalkTests.cs
+194
View File
@@ -0,0 +1,194 @@
using Opc.Ua;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using HistorianRead = ZB.MOM.WW.OtOpcUa.Core.Abstractions.HistoryReadResult;
namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests;
/// <summary>
/// Gitea #689 / #691: the reverse walk-back (<see cref="ReverseWalk"/> through <see cref="ReverseRawReader"/>) driven
/// directly, without a server: a randomized exactness check against a brute-force answer, and the sizing properties.
/// </summary>
public sealed class ReverseWalkTests
{
private static CancellationToken Ct => TestContext.Current.CancellationToken;
private static readonly DateTime T0 = new(2026, 10, 12, 18, 0, 0, DateTimeKind.Utc);
private static DataValueSnapshot At(long ticks, double v) =>
new(v, StatusCodes.Good, new DateTime(ticks, DateTimeKind.Utc), new DateTime(ticks, DateTimeKind.Utc));
/// <summary>
/// A randomized walk over AVEVA-shaped forward reads (inclusive end, a fabricated row at the query start — optionally
/// stamped at the start rounded UP to the millisecond — counted against the cap): every page is exactly the brute-force
/// newest-first answer over <c>(lo, hi]</c>, ties in the backend's forward order reversed, whatever the carried width
/// — and a timestamp over the tie budget is the only refusal.
/// </summary>
[Fact]
public async Task A_walked_page_is_exactly_the_newest_N_over_the_window_on_random_data()
{
var failures = new List<string>();
for (var seed = 0; seed < 400; seed++)
{
var rng = new Random(seed);
var series = RandomSeries(rng, out var maxCluster);
var source = new AvevaShapedSource(series, roundUp: rng.Next(2) == 0);
var tieBudget = rng.Next(4) switch { 0 => 1, 1 => 3, 2 => 8, _ => 65536 };
var want = (uint)rng.Next(1, 40);
var span = Ts(series[^1]) - Ts(series[0]);
var hi = Ts(series[0]) + (long)(rng.NextDouble() * 1.2 * span) + rng.Next(0, 3);
var lo = Ts(series[0]) - TimeSpan.TicksPerSecond + (long)(rng.NextDouble() * (hi - Ts(series[0])) * 0.8);
if (rng.Next(5) == 0) lo = 0; // an end-only read
var hint = rng.Next(3) switch
{
0 => default,
1 => new ReverseWalkHint(rng.Next(1, 10)),
_ => new ReverseWalkHint((long)(rng.NextDouble() * TimeSpan.TicksPerHour)),
};
var expected = series.Where(s => Ts(s) > Math.Max(lo, DateTime.UnixEpoch.Ticks) && Ts(s) <= hi)
.Select((s, i) => (s, i)).OrderByDescending(x => Ts(x.s)).ThenByDescending(x => x.i)
.Take((int)want).Select(x => (double)x.s.Value!).ToList();
try
{
var page = await ReverseRawReader.ReadAsync(source, "T", new DateTime(hi, DateTimeKind.Utc),
new DateTime(lo, DateTimeKind.Utc), want, tieBudget, T0.AddDays(1), hint, Ct);
var got = page.Samples.Select(s => (double)s.Value!).ToList();
if (!got.SequenceEqual(expected))
failures.Add($"seed {seed}: got [{string.Join(",", got)}] expected [{string.Join(",", expected)}]");
if (page.Hint.WidthTicks <= 0) failures.Add($"seed {seed}: no width carried");
}
catch (HistoryReadLimitExceededException)
{
if (maxCluster <= tieBudget) failures.Add($"seed {seed}: refused with no cluster over {tieBudget} (max {maxCluster})");
}
if (source.MaxCap > 2L * Math.Max(Math.Max(tieBudget, ReverseWalk.DiscoveryRows), 2 * want + 16) + 4)
failures.Add($"seed {seed}: asked for {source.MaxCap} rows");
if (source.BadRequests > 0) failures.Add($"seed {seed}: {source.BadRequests} invalid requests");
}
failures.ShouldBeEmpty();
}
/// <summary>
/// #689: the carried width sizes a resumed page from N, not from the tie budget — the live symptom was 32,770 rows
/// per page for N = 3 on a 1 Hz tag once the width had grown over a few pages.
/// </summary>
[Fact]
public async Task A_resumed_page_on_a_dense_tag_reads_about_N_rows_with_the_default_tie_budget()
{
var series = Enumerable.Range(0, 200_000).Select(i => At(T0.Ticks + i * TimeSpan.TicksPerSecond, i)).ToList();
var source = new AvevaShapedSource(series);
var hi = new DateTime(Ts(series[^1]), DateTimeKind.Utc);
var hint = default(ReverseWalkHint);
for (var page = 1; page <= 30; page++)
{
source.Reset();
var result = await ReverseRawReader.ReadAsync(source, "T", hi, DateTime.MinValue, 3, 65536, hi.AddHours(1), hint, Ct);
result.Samples.Count.ShouldBe(3);
hint = result.Hint;
hi = result.Samples[^1].SourceTimestampUtc!.Value.AddTicks(-1);
if (page > 1) source.RowsReturned.ShouldBeLessThanOrEqualTo(30, $"page {page}");
}
}
/// <summary>
/// #689: a dense tag that stopped long ago. The walk's ×4 growth across the empty stretch overshoots into far more
/// rows than one read takes; the truncated read only shows the OLDEST of them, so the walk must not keep re-entering
/// that stretch — it bisects the gap above the newest row it saw.
/// </summary>
[Fact]
public async Task A_dense_tag_below_a_long_empty_stretch_is_found_in_a_bounded_number_of_reads()
{
// 10 Hz for 3 hours, ending two days before "now".
var start = T0.AddDays(-2).AddHours(-3);
var series = Enumerable.Range(0, 108_000).Select(i => At(start.Ticks + i * TimeSpan.TicksPerMillisecond * 100, i)).ToList();
var source = new AvevaShapedSource(series);
var page = await ReverseRawReader.ReadAsync(source, "T", T0, DateTime.MinValue, 3, 65536, T0, default(ReverseWalkHint), Ct);
page.Samples.Select(s => (double)s.Value!).ShouldBe([107_999d, 107_998d, 107_997d]);
source.Reads.ShouldBeLessThanOrEqualTo(25);
source.RowsReturned.ShouldBeLessThanOrEqualTo(25 * (ReverseWalk.DiscoveryRows + 2));
}
private static long Ts(DataValueSnapshot s) => s.SourceTimestampUtc!.Value.Ticks;
/// <summary>Random ascending series: bursts, gaps, sub-millisecond stamps and tie clusters.</summary>
private static List<DataValueSnapshot> RandomSeries(Random rng, out int maxCluster)
{
var list = new List<DataValueSnapshot>();
var t = T0.Ticks - rng.Next(0, 3) * TimeSpan.TicksPerDay;
var count = rng.Next(1, 400);
maxCluster = 1;
var v = 0;
while (list.Count < count)
{
t += rng.Next(6) switch
{
0 => rng.Next(1, 10), // sub-millisecond
1 => rng.Next(1, 5) * TimeSpan.TicksPerMillisecond, // dense
2 => rng.Next(1, 3) * TimeSpan.TicksPerSecond,
3 => rng.Next(1, 120) * TimeSpan.TicksPerMinute, // a gap
_ => rng.Next(1, 1000) * TimeSpan.TicksPerMillisecond,
};
var ties = rng.Next(8) == 0 ? rng.Next(2, 12) : 1;
maxCluster = Math.Max(maxCluster, ties);
for (var k = 0; k < ties; k++) list.Add(At(t, v++));
}
return list;
}
/// <summary>AVEVA-shaped forward-only raw source (see the fuzz test's summary).</summary>
private sealed class AvevaShapedSource(IReadOnlyList<DataValueSnapshot> series, bool roundUp = false) : IHistorianDataSource
{
public long MaxCap { get; private set; }
public long RowsReturned { get; private set; }
public int Reads { get; private set; }
public int BadRequests { get; private set; }
public void Reset() { MaxCap = 0; RowsReturned = 0; Reads = 0; }
public bool SupportsReverseRaw => false;
public Task<HistorianRead> ReadRawAsync(string fullReference, DateTime startUtc, DateTime endUtc, uint maxValuesPerNode,
CancellationToken cancellationToken)
{
Reads++;
if (startUtc > endUtc || maxValuesPerNode == 0 || startUtc < DateTime.UnixEpoch) BadRequests++;
MaxCap = Math.Max(MaxCap, maxValuesPerNode);
var stamp = roundUp && startUtc.Ticks % TimeSpan.TicksPerMillisecond != 0
? new DateTime(startUtc.Ticks - startUtc.Ticks % TimeSpan.TicksPerMillisecond + TimeSpan.TicksPerMillisecond, DateTimeKind.Utc)
: startUtc;
var prior = series.LastOrDefault(s => s.SourceTimestampUtc <= startUtc);
var rows = new List<DataValueSnapshot>
{
prior is null
? new DataValueSnapshot(-1d, StatusCodes.BadNoData, stamp, stamp)
: new DataValueSnapshot(-2d, prior.StatusCode, stamp, stamp),
};
rows.AddRange(series.Where(s => s.SourceTimestampUtc >= startUtc && s.SourceTimestampUtc <= endUtc));
var page = rows.Take((int)Math.Min(int.MaxValue, maxValuesPerNode)).ToList();
RowsReturned += page.Count;
return Task.FromResult(new HistorianRead(page, null));
}
public Task<HistorianRead> ReadProcessedAsync(string fullReference, DateTime startUtc, DateTime endUtc, TimeSpan interval,
HistoryAggregateType aggregate, CancellationToken cancellationToken) => throw new NotSupportedException();
public Task<HistorianRead> ReadAtTimeAsync(string fullReference, IReadOnlyList<DateTime> timestampsUtc,
CancellationToken cancellationToken) => throw new NotSupportedException();
public Task<HistoricalEventsResult> ReadEventsAsync(string? sourceName, DateTime startUtc, DateTime endUtc, int maxEvents,
CancellationToken cancellationToken) => throw new NotSupportedException();
public HistorianHealthSnapshot GetHealthSnapshot() => NullHistorianDataSource.Instance.GetHealthSnapshot();
public void Dispose() { }
}
}