fix(historian): #694 page raw reads at a strict gateway's learned ceiling; opt-in EventReadHorizon
Item 1: IHistorianDataSource.MaxRawValuesPerRead (Gateway: learned ceiling - 16 = L; 0 elsewhere, guard-tested). RawPageLimit plans every raw page on top of #692's lookahead: with L > 0 the read (page P + boundary ties k + lookahead) is <= L, P = min(N, L - k - 1) (uncapped: L - k - 1); no room => straight to the tie-cluster over-fetch (never an uncapped or underflowing request). One re-plan + re-read when the read taught a new limit or was refused for crowding pre-window rows (HistoryReadLimitExceededException.PreWindowRows): the page shrinks by them. Over-fetch bound min(MaxTieClusterOverfetch, L - 1); slice size never 0. Review fixes: F1 a window starting at/before the epoch counts AVEVA's synthetic row as pre-window (a ceiling-truncated read was reported complete); F2 the page path disables ReverseRawReader's own re-run (cold end-only ceiling 20/N=3000 cost 3,023 reads, now <= 40); F3 crowded look-back re-read around, not refused; F4 the #687 crowding re-read room at least doubles (up to 8 re-reads), so a cluster of hundreds just before a resume point no longer fails on a default gateway; a crowding refusal that leaves no room is BadTooManyOperations, never a tick-by-tick cursor. Item 2: ServerHistorian:EventReadHorizon (default 00:00:00 = exact, unchanged). When set, every fresh HistoryReadEvents window's earlier bound is raised to now - horizon (both directions, all backends). Lossy by design; documented. Item 3 (native TimeAverage/Total re-check) stays blocked on historiangw#27.
This commit is contained in:
+15
@@ -94,4 +94,19 @@ public sealed class HistorianDataSourceReverseEventsGuardTests
|
||||
$"{type.FullName} must declare SupportedAggregates itself, not inherit the interface default");
|
||||
}
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Every_implementation_declares_its_raw_page_limit_itself()
|
||||
{
|
||||
// Gitea #694: the default (0 = no limit) is what a strict-ceiling Gateway must NOT report — a wrapper that inherited
|
||||
// it would bring back BadTooManyOperations for every page larger than the gateway's ceiling.
|
||||
foreach (var type in ProductionImplementations())
|
||||
{
|
||||
var map = type.GetInterfaceMap(typeof(IHistorianDataSource));
|
||||
var getter = map.TargetMethods[Array.FindIndex(map.InterfaceMethods,
|
||||
m => m.Name == "get_" + nameof(IHistorianDataSource.MaxRawValuesPerRead))];
|
||||
getter.DeclaringType.ShouldBe(type,
|
||||
$"{type.FullName} must declare MaxRawValuesPerRead itself, not inherit the interface default");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
+499
@@ -0,0 +1,499 @@
|
||||
using Google.Protobuf.WellKnownTypes;
|
||||
using Microsoft.Extensions.Logging.Abstractions;
|
||||
using Opc.Ua;
|
||||
using Opc.Ua.Server;
|
||||
using Shouldly;
|
||||
using Xunit;
|
||||
using ZB.MOM.WW.HistorianGateway.Contracts.Grpc;
|
||||
using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
|
||||
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
||||
using ZB.MOM.WW.OtOpcUa.OpcUaServer;
|
||||
using SdkHistoryReadResult = Opc.Ua.HistoryReadResult;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests;
|
||||
|
||||
/// <summary>
|
||||
/// Gitea #694 item 1 (live gate F3): a strict gateway (<c>AllowStreamingBeyondCeiling=false</c>) with a LOWERED
|
||||
/// <c>MaxValuesCeiling</c> (1000) refused ordinary page sizes — <c>NumValuesPerNode = 5000</c>, or an uncapped read over
|
||||
/// 3,600 rows, came back <c>BadTooManyOperations</c>. The server now sizes the page to the learned ceiling
|
||||
/// (<see cref="IHistorianDataSource.MaxRawValuesPerRead"/> L = ceiling − 16): the read — page P + boundary ties k + the
|
||||
/// #692 lookahead row — is at most L, so a first page holds L − 1 values and a resumed one L − k − 1, each with a
|
||||
/// continuation point. Driven end to end: a REAL SDK server's node manager → the REAL
|
||||
/// <see cref="GatewayHistorianDataSource"/> → an AVEVA-shaped fake gateway client that refuses an over-ceiling
|
||||
/// <c>max_values</c> exactly as the real gateway does (<c>InvalidArgument</c> "max_values exceeds the server ceiling of N",
|
||||
/// wrapped by the client package) and rejects <c>max_values <= 0</c> as the client package does.
|
||||
/// </summary>
|
||||
public sealed class GatewayCeilingPagingTests : IDisposable
|
||||
{
|
||||
private static CancellationToken Ct => TestContext.Current.CancellationToken;
|
||||
|
||||
private static readonly DateTime T0 = new(2026, 10, 12, 10, 0, 0, DateTimeKind.Utc);
|
||||
|
||||
private const int Ceiling = 1000;
|
||||
private const int L = Ceiling - RawAggregateReader.PreWindowAllowance; // 984
|
||||
private const int FirstPage = L - 1; // 983: page + lookahead fills a read
|
||||
private const int NextPage = L - 2; // 982: + the one boundary tie a resumed page re-reads (distinct timestamps)
|
||||
|
||||
private readonly string _pkiRoot = Path.Combine(Path.GetTempPath(), $"otopcua-gwceiling-{Guid.NewGuid():N}");
|
||||
|
||||
private static DateTime At(int seconds) => T0.AddSeconds(seconds);
|
||||
|
||||
private static HistorianSample Row(DateTime t, double v, uint quality = 0) =>
|
||||
new() { Tag = "T", Timestamp = Timestamp.FromDateTime(t), NumericValue = v, OpcQuality = 192, Quality = quality };
|
||||
|
||||
/// <summary>
|
||||
/// AVEVA-shaped ReadRaw over <paramref name="stored"/>: a boundary row at the query start (the value in effect, or a
|
||||
/// NULL row before the first sample), then <c>[start, end]</c> inclusive ascending, capped at <c>max_values</c>.
|
||||
/// </summary>
|
||||
private static FakeHistorianGatewayClient Gateway(IReadOnlyList<(DateTime T, double V)> stored, int? ceiling) => new()
|
||||
{
|
||||
MaxValuesCeiling = ceiling,
|
||||
RawSource = (start, end, max) =>
|
||||
{
|
||||
if (max <= 0) throw new ArgumentOutOfRangeException(nameof(max), "max_values must be positive"); // the client package
|
||||
var rows = new List<HistorianSample>();
|
||||
if (stored.All(s => s.T != start))
|
||||
{
|
||||
var prior = stored.LastOrDefault(s => s.T < start);
|
||||
rows.Add(prior.T == default
|
||||
? new HistorianSample { Tag = "T", Timestamp = Timestamp.FromDateTime(start), NumericValue = 0, OpcQuality = 0, Quality = 1 }
|
||||
: Row(start, prior.V, 133));
|
||||
}
|
||||
rows.AddRange(stored.Where(s => s.T >= start && s.T <= end).Select(s => Row(s.T, s.V)));
|
||||
return rows.Take(max);
|
||||
},
|
||||
};
|
||||
|
||||
/// <summary>One sample per second, value = seconds, at +0 … +(count − 1).</summary>
|
||||
private static List<(DateTime T, double V)> PerSecond(int count) =>
|
||||
Enumerable.Range(0, count).Select(i => (At(i), (double)i)).ToList();
|
||||
|
||||
private static int Refusals(FakeHistorianGatewayClient fake) =>
|
||||
fake.RawCalls.Count(c => fake.MaxValuesCeiling is { } n && c.MaxValues > n);
|
||||
|
||||
private static void EveryRequestFits(FakeHistorianGatewayClient fake) =>
|
||||
fake.RawCalls.ShouldAllBe(c => c.MaxValues > 0 && c.MaxValues <= Ceiling); // never uncapped, never over the ceiling
|
||||
|
||||
// ---- the data source's page limit ------------------------------------------------------------------------------------
|
||||
|
||||
[Fact]
|
||||
public async Task The_page_limit_is_unknown_until_a_refusal_names_the_ceiling_and_forgotten_when_the_gateway_streams()
|
||||
{
|
||||
var fake = Gateway(PerSecond(3600), Ceiling);
|
||||
var ds = new GatewayHistorianDataSource(fake, NullLogger<GatewayHistorianDataSource>.Instance);
|
||||
ds.MaxRawValuesPerRead.ShouldBe(0);
|
||||
|
||||
await ds.ReadRawAsync("T", At(0), At(10), 0, Ct);
|
||||
ds.MaxRawValuesPerRead.ShouldBe(L); // a read of this size asks for exactly the ceiling
|
||||
|
||||
(await ds.ReadRawAsync("T", At(0), At(3600), (uint)L, Ct)).Samples.Count.ShouldBe(L);
|
||||
fake.RawCalls[^1].MaxValues.ShouldBe(Ceiling);
|
||||
Refusals(fake).ShouldBe(1);
|
||||
|
||||
// The operator enabled AllowStreamingBeyondCeiling: a ceiling-full uncapped read re-probes (#691.3) and forgets it.
|
||||
fake.MaxValuesCeiling = null;
|
||||
(await ds.ReadRawAsync("T", At(0), At(3600), 0, Ct)).Samples.Count.ShouldBe(3600);
|
||||
ds.MaxRawValuesPerRead.ShouldBe(0);
|
||||
}
|
||||
|
||||
// ---- through the node manager ----------------------------------------------------------------------------------------
|
||||
|
||||
[Fact]
|
||||
public async Task The_cold_first_read_costs_one_refusal_and_its_page_fits_the_learned_ceiling()
|
||||
{
|
||||
var fake = Gateway(PerSecond(3600), Ceiling);
|
||||
var (host, nm, nodeId) = await BootAsync(fake);
|
||||
try
|
||||
{
|
||||
var first = ReadRaw(nm, nodeId, At(0), At(3600), 5000, null);
|
||||
first.Error.StatusCode.Code.ShouldBe(StatusCodes.Good);
|
||||
first.Values.Count.ShouldBe(FirstPage);
|
||||
first.Cp.ShouldNotBeNull();
|
||||
// The budget the client asked for (+ lookahead + allowance, refused naming 1000), the ceiling re-read that cannot
|
||||
// hold the window, then the page sized to the learned ceiling.
|
||||
fake.RawCalls.Select(c => c.MaxValues).ShouldBe([5000 + 1 + RawAggregateReader.PreWindowAllowance, Ceiling, Ceiling]);
|
||||
|
||||
var (rest, _, status) = DrainFrom(nm, nodeId, At(0), At(3600), 5000, first.Cp);
|
||||
status.Code.ShouldBe(StatusCodes.Good);
|
||||
first.Values.Concat(rest).ShouldBe(PerSecond(3600).Select(s => s.Item2));
|
||||
Refusals(fake).ShouldBe(1);
|
||||
}
|
||||
finally { await host.DisposeAsync(); }
|
||||
}
|
||||
|
||||
[Theory]
|
||||
[InlineData(5000u, false)]
|
||||
[InlineData(0u, false)]
|
||||
[InlineData(5000u, true)]
|
||||
[InlineData(0u, true)]
|
||||
public async Task With_the_ceiling_known_every_page_is_exactly_the_ceiling_page_and_nothing_is_refused(uint n, bool reverse)
|
||||
{
|
||||
var stored = PerSecond(3600);
|
||||
var fake = Gateway(stored, Ceiling);
|
||||
var (host, nm, nodeId) = await BootAsync(fake);
|
||||
try
|
||||
{
|
||||
ReadRaw(nm, nodeId, At(0), At(3600), 5000, null); // learns the ceiling
|
||||
fake.RawCalls.Clear();
|
||||
|
||||
var (values, pages, status) = reverse
|
||||
? DrainFrom(nm, nodeId, At(3600), At(-1), n, null)
|
||||
: DrainFrom(nm, nodeId, At(0), At(3600), n, null);
|
||||
|
||||
status.Code.ShouldBe(StatusCodes.Good);
|
||||
var expected = stored.Select(s => s.Item2);
|
||||
values.ShouldBe(reverse ? expected.Reverse() : expected);
|
||||
pages[0].ShouldBe(FirstPage);
|
||||
pages.Skip(1).SkipLast(1).ShouldAllBe(p => p == NextPage);
|
||||
pages[^1].ShouldBeGreaterThan(0); // no empty trailing page
|
||||
Refusals(fake).ShouldBe(0);
|
||||
EveryRequestFits(fake);
|
||||
if (!reverse) fake.RawCalls.Count.ShouldBe(pages.Count); // one read per page
|
||||
}
|
||||
finally { await host.DisposeAsync(); }
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task An_uncapped_read_that_fits_the_ceiling_is_still_one_page_without_a_continuation_point()
|
||||
{
|
||||
var fake = Gateway(PerSecond(3600), Ceiling);
|
||||
var (host, nm, nodeId) = await BootAsync(fake);
|
||||
try
|
||||
{
|
||||
var (values, pages, _) = DrainFrom(nm, nodeId, At(0), At(10), 0, null);
|
||||
values.Count.ShouldBe(10);
|
||||
pages.Count.ShouldBe(1);
|
||||
fake.RawCalls.Select(c => c.MaxValues).ShouldBe([int.MaxValue, Ceiling]); // #691.3 unchanged
|
||||
|
||||
var again = DrainFrom(nm, nodeId, At(0), At(10), 0, null);
|
||||
again.Pages.Count.ShouldBe(1);
|
||||
fake.RawCalls[^1].MaxValues.ShouldBe(Ceiling); // the learned page size, one read
|
||||
}
|
||||
finally { await host.DisposeAsync(); }
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task A_default_gateway_is_unchanged_one_read_and_no_continuation_point()
|
||||
{
|
||||
var fake = Gateway(PerSecond(3600), ceiling: null);
|
||||
var (host, nm, nodeId) = await BootAsync(fake);
|
||||
try
|
||||
{
|
||||
var (values, pages, _) = DrainFrom(nm, nodeId, At(0), At(3600), 5000, null);
|
||||
values.Count.ShouldBe(3600);
|
||||
pages.Count.ShouldBe(1);
|
||||
fake.RawCalls.Select(c => c.MaxValues).ShouldBe([5000 + 1 + RawAggregateReader.PreWindowAllowance]);
|
||||
}
|
||||
finally { await host.DisposeAsync(); }
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// A tie cluster straddling a page boundary, at the sizes where the ceiling arithmetic turns: a read can carry L rows;
|
||||
/// the over-fetch that pages within a cluster reads at most L − 1 (+1 to detect more). Every outcome is exact, or
|
||||
/// <c>BadTooManyOperations</c> after an exact prefix — never an uncapped or over-ceiling request, never an endless read.
|
||||
/// </summary>
|
||||
[Theory]
|
||||
[InlineData(L - 2, 5000u)]
|
||||
[InlineData(L - 1, 5000u)]
|
||||
[InlineData(L, 5000u)]
|
||||
[InlineData(L + 1, 5000u)]
|
||||
[InlineData(Ceiling - 17, 5000u)]
|
||||
[InlineData(Ceiling - 16, 5000u)]
|
||||
[InlineData(Ceiling + 1, 5000u)]
|
||||
[InlineData(L - 2, 0u)]
|
||||
[InlineData(L - 1, 0u)]
|
||||
[InlineData(L + 1, 0u)]
|
||||
[InlineData(Ceiling + 1, 0u)]
|
||||
public async Task A_tie_cluster_at_a_page_boundary_is_exact_or_refused_never_truncated_or_endless(int size, uint n)
|
||||
{
|
||||
var stored = PerSecond(500)
|
||||
.Concat(Enumerable.Range(0, size).Select(i => (At(600), 10_000.0 + i)))
|
||||
.Concat(Enumerable.Range(0, 1500).Select(i => (At(700 + i), 20_000.0 + i)))
|
||||
.ToList();
|
||||
var fake = Gateway(stored, Ceiling);
|
||||
var (host, nm, nodeId) = await BootAsync(fake);
|
||||
try
|
||||
{
|
||||
ReadRaw(nm, nodeId, At(0), At(5), 5000, null); // learns the ceiling
|
||||
fake.RawCalls.Clear();
|
||||
|
||||
var (values, pages, status) = DrainFrom(nm, nodeId, At(0), At(5000), n, null, maxPages: 60);
|
||||
var expected = stored.Select(s => s.Item2).ToList();
|
||||
if (status.Code == StatusCodes.Good) values.ShouldBe(expected);
|
||||
else
|
||||
{
|
||||
status.Code.ShouldBe(StatusCodes.BadTooManyOperations);
|
||||
values.ShouldBe(expected.Take(values.Count)); // an exact prefix
|
||||
size.ShouldBeGreaterThan(L - 2, "a cluster that fits the over-fetch is always served");
|
||||
}
|
||||
pages.Count.ShouldBeLessThan(60);
|
||||
EveryRequestFits(fake);
|
||||
}
|
||||
finally { await host.DisposeAsync(); }
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task An_uncapped_cursor_inside_a_tie_cluster_finishes_exactly_when_the_gateway_stops_being_strict()
|
||||
{
|
||||
// 100 samples, then 983 ties: the first page ends inside the cluster. Then the operator enables streaming and the
|
||||
// data source forgets the ceiling (L back to 0) — the uncapped cursor must still finish exactly, not loop.
|
||||
var stored = PerSecond(100)
|
||||
.Concat(Enumerable.Range(0, L - 1).Select(i => (At(200), 10_000.0 + i)))
|
||||
.Concat(Enumerable.Range(0, 300).Select(i => (At(300 + i), 20_000.0 + i)))
|
||||
.ToList();
|
||||
var fake = Gateway(stored, Ceiling);
|
||||
var ds = new GatewayHistorianDataSource(fake, NullLogger<GatewayHistorianDataSource>.Instance);
|
||||
var (host, nm, nodeId) = await BootAsync(fake, ds);
|
||||
try
|
||||
{
|
||||
ReadRaw(nm, nodeId, At(0), At(5), 5000, null); // learns the ceiling
|
||||
var first = ReadRaw(nm, nodeId, At(0), At(5000), 0, null);
|
||||
first.Values.Count.ShouldBe(FirstPage);
|
||||
first.Cp.ShouldNotBeNull();
|
||||
|
||||
fake.MaxValuesCeiling = null;
|
||||
await ds.ReadRawAsync("T", At(0), At(5000), 0, Ct); // a ceiling-full read re-probes and forgets
|
||||
ds.MaxRawValuesPerRead.ShouldBe(0);
|
||||
|
||||
var (rest, pages, status) = DrainFrom(nm, nodeId, At(0), At(5000), 0, first.Cp, maxPages: 20);
|
||||
status.Code.ShouldBe(StatusCodes.Good);
|
||||
first.Values.Concat(rest).ShouldBe(stored.Select(s => s.Item2));
|
||||
pages.Count.ShouldBeLessThan(20);
|
||||
}
|
||||
finally { await host.DisposeAsync(); }
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task A_ceiling_lowered_mid_page_through_is_re_learned_and_the_read_stays_exact()
|
||||
{
|
||||
var stored = PerSecond(3600);
|
||||
var fake = Gateway(stored, Ceiling);
|
||||
var (host, nm, nodeId) = await BootAsync(fake);
|
||||
try
|
||||
{
|
||||
var got = new List<double>();
|
||||
var p1 = ReadRaw(nm, nodeId, At(0), At(3600), 5000, null);
|
||||
got.AddRange(p1.Values);
|
||||
var p2 = ReadRaw(nm, nodeId, At(0), At(3600), 5000, p1.Cp);
|
||||
got.AddRange(p2.Values);
|
||||
|
||||
fake.MaxValuesCeiling = 500; // the operator lowered it
|
||||
fake.RawCalls.Clear();
|
||||
var (rest, pages, status) = DrainFrom(nm, nodeId, At(0), At(3600), 5000, p2.Cp, maxPages: 20);
|
||||
status.Code.ShouldBe(StatusCodes.Good);
|
||||
got.AddRange(rest);
|
||||
got.ShouldBe(stored.Select(s => s.Item2));
|
||||
pages.ShouldAllBe(p => p <= 500 - RawAggregateReader.PreWindowAllowance);
|
||||
Refusals(fake).ShouldBe(1); // the one that names the new ceiling
|
||||
}
|
||||
finally { await host.DisposeAsync(); }
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Bounding_values_page_within_the_ceiling()
|
||||
{
|
||||
var stored = PerSecond(3600);
|
||||
var fake = Gateway(stored, Ceiling);
|
||||
var (host, nm, nodeId) = await BootAsync(fake);
|
||||
try
|
||||
{
|
||||
ReadRaw(nm, nodeId, At(0), At(5), 5000, null); // learns the ceiling
|
||||
fake.RawCalls.Clear();
|
||||
|
||||
// Start bound = the sample at +10 (before StartTime), end bound = the sample at +3001 (after EndTime).
|
||||
var start = At(10).AddMilliseconds(500);
|
||||
var end = At(3000).AddMilliseconds(500);
|
||||
var (values, pages, status) = DrainFrom(nm, nodeId, start, end, 5000, null, returnBounds: true);
|
||||
status.Code.ShouldBe(StatusCodes.Good);
|
||||
values.ShouldBe(Enumerable.Range(10, 2992).Select(i => (double)i));
|
||||
pages.ShouldAllBe(p => p <= L);
|
||||
Refusals(fake).ShouldBe(0);
|
||||
}
|
||||
finally { await host.DisposeAsync(); }
|
||||
}
|
||||
|
||||
[Theory]
|
||||
[InlineData(1000, 100_000u, 10)]
|
||||
[InlineData(1000, 5000u, 10)]
|
||||
[InlineData(20, 3000u, 40)]
|
||||
public async Task A_cold_end_only_read_with_a_huge_page_shrinks_the_page_not_the_walk(int ceiling, uint n, int maxReads)
|
||||
{
|
||||
// Review F2: the walk used to re-run itself at the learned ceiling while keeping the page at N — a cold end-only read
|
||||
// with ceiling 20 and N = 3000 cost 3,023 gateway reads in ONE HistoryRead (ceiling 1000, N = 100,000: 207). The
|
||||
// refusal now reaches the page planner, which shrinks the page to the ceiling.
|
||||
var dense = Enumerable.Range(0, 200_000).Select(i => (T0.AddMilliseconds(i), (double)i)).ToList();
|
||||
var fake = Gateway(dense, ceiling);
|
||||
var (host, nm, nodeId) = await BootAsync(fake);
|
||||
try
|
||||
{
|
||||
var r = ReadRaw(nm, nodeId, DateTime.MinValue, T0.AddMilliseconds(200_000), n, null);
|
||||
r.Error.StatusCode.Code.ShouldBe(StatusCodes.Good);
|
||||
r.Cp.ShouldNotBeNull();
|
||||
var page = ceiling - RawAggregateReader.PreWindowAllowance - 1;
|
||||
r.Values.ShouldBe(Enumerable.Range(200_000 - r.Values.Count, r.Values.Count).Reverse().Select(i => (double)i));
|
||||
r.Values.Count.ShouldBeLessThanOrEqualTo(page);
|
||||
fake.RawCalls.Count.ShouldBeLessThanOrEqualTo(maxReads);
|
||||
}
|
||||
finally { await host.DisposeAsync(); }
|
||||
}
|
||||
|
||||
// ---- F1: a window starting at/before the epoch gets no look-back (review of #694) -------------------------------------
|
||||
|
||||
private static List<(DateTime T, double V)> FromEpoch(int count) =>
|
||||
Enumerable.Range(1, count).Select(i => (DateTime.UnixEpoch.AddSeconds(i), (double)i)).ToList();
|
||||
|
||||
[Fact]
|
||||
public async Task A_ceiling_truncated_read_from_the_epoch_is_never_reported_complete()
|
||||
{
|
||||
// AVEVA's synthetic row lands AT the epoch start (no 1 ms look-back there). It used to count as "past the window
|
||||
// end", so a ceiling-truncated uncapped read answered its first 999 rows, Good, as if that were everything.
|
||||
var fake = Gateway(FromEpoch(3000), Ceiling);
|
||||
var ds = new GatewayHistorianDataSource(fake, NullLogger<GatewayHistorianDataSource>.Instance);
|
||||
await Should.ThrowAsync<HistoryReadLimitExceededException>(() =>
|
||||
ds.ReadRawAsync("T", DateTime.UnixEpoch, DateTime.UnixEpoch.AddDays(1), 0, Ct));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task An_uncapped_read_from_the_epoch_pages_every_sample()
|
||||
{
|
||||
var stored = FromEpoch(3000);
|
||||
var fake = Gateway(stored, Ceiling);
|
||||
var (host, nm, nodeId) = await BootAsync(fake);
|
||||
try
|
||||
{
|
||||
var (values, pages, status) = DrainFrom(nm, nodeId, DateTime.UnixEpoch, DateTime.UnixEpoch.AddDays(1), 0, null);
|
||||
status.Code.ShouldBe(StatusCodes.Good);
|
||||
values.ShouldBe(stored.Select(s => s.Item2));
|
||||
pages.Count.ShouldBeGreaterThan(1);
|
||||
}
|
||||
finally { await host.DisposeAsync(); }
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task An_end_only_read_far_past_epoch_data_returns_the_latest_values()
|
||||
{
|
||||
var stored = FromEpoch(3000);
|
||||
var fake = Gateway(stored, Ceiling);
|
||||
var (host, nm, nodeId) = await BootAsync(fake);
|
||||
try
|
||||
{
|
||||
ReadRaw(nm, nodeId, DateTime.UnixEpoch, DateTime.UnixEpoch.AddSeconds(5), 5000, null); // learns the ceiling
|
||||
var r = ReadRaw(nm, nodeId, DateTime.MinValue, DateTime.UnixEpoch.AddDays(30), 10, null);
|
||||
r.Error.StatusCode.Code.ShouldBe(StatusCodes.Good);
|
||||
r.Values.ShouldBe(Enumerable.Range(2991, 10).Reverse().Select(i => (double)i)); // the LATEST ten, newest first
|
||||
}
|
||||
finally { await host.DisposeAsync(); }
|
||||
}
|
||||
|
||||
// ---- F3: rows crowding the look-back before a page start ---------------------------------------------------------------
|
||||
|
||||
[Theory]
|
||||
[InlineData(5000u)]
|
||||
[InlineData(0u)]
|
||||
[InlineData((uint)L)]
|
||||
[InlineData(900u)]
|
||||
public async Task Ties_just_before_a_page_start_are_read_around_not_refused(uint n)
|
||||
{
|
||||
// 962 samples, 20 ties at X, one sample at X + 1 ms: the first page (983) ends exactly at X + 1 ms. The next page's
|
||||
// 1 ms look-back holds the 20 ties — more than the 16-row allowance, and a ceiling-full page leaves no room for the
|
||||
// #687 re-read, so it used to fail BadTooManyOperations. The source now names the crowding rows and the page is
|
||||
// re-read that much smaller.
|
||||
var x = At(1000);
|
||||
var stored = Enumerable.Range(0, 962).Select(i => (At(i), (double)i))
|
||||
.Concat(Enumerable.Range(0, 20).Select(i => (x, 962.0 + i)))
|
||||
.Append((x.AddMilliseconds(1), 982.0))
|
||||
.Concat(Enumerable.Range(0, 3000).Select(i => (x.AddSeconds(1 + i), 983.0 + i)))
|
||||
.ToList();
|
||||
var fake = Gateway(stored, Ceiling);
|
||||
var (host, nm, nodeId) = await BootAsync(fake);
|
||||
try
|
||||
{
|
||||
ReadRaw(nm, nodeId, At(0), At(5), 5000, null); // learns the ceiling
|
||||
fake.RawCalls.Clear();
|
||||
|
||||
var (values, _, status) = DrainFrom(nm, nodeId, At(0), x.AddDays(1), n, null);
|
||||
status.Code.ShouldBe(StatusCodes.Good);
|
||||
values.ShouldBe(stored.Select(s => s.Item2));
|
||||
Refusals(fake).ShouldBe(0);
|
||||
EveryRequestFits(fake);
|
||||
}
|
||||
finally { await host.DisposeAsync(); }
|
||||
}
|
||||
|
||||
// ---- helpers ---------------------------------------------------------------------------------------------------------
|
||||
|
||||
private async Task<(OpcUaApplicationHost Host, OtOpcUaNodeManager Nm, NodeId NodeId)> BootAsync(
|
||||
FakeHistorianGatewayClient fake, GatewayHistorianDataSource? ds = null)
|
||||
{
|
||||
var started = await OpcUaHostStart.StartAsync(
|
||||
port => new OpcUaApplicationHost(
|
||||
new OpcUaApplicationHostOptions
|
||||
{
|
||||
ApplicationName = "OtOpcUa.GatewayCeilingPagingTest",
|
||||
ApplicationUri = $"urn:OtOpcUa.GatewayCeilingPagingTest:{Guid.NewGuid():N}",
|
||||
OpcUaPort = port,
|
||||
PublicHostname = "localhost",
|
||||
PkiStoreRoot = _pkiRoot,
|
||||
},
|
||||
NullLogger<OpcUaApplicationHost>.Instance),
|
||||
() => new OtOpcUaSdkServer(), Ct);
|
||||
var nm = started.Server.NodeManager!;
|
||||
nm.HistoryContinuationStore = new InMemoryHistoryContinuationStore();
|
||||
nm.HistorianDataSource = ds ?? new GatewayHistorianDataSource(fake, NullLogger<GatewayHistorianDataSource>.Instance);
|
||||
nm.EnsureVariable("eq-1/pv", parentFolderNodeId: null, displayName: "PV", dataType: "Double",
|
||||
writable: false, historianTagname: "T", realm: AddressSpaceRealm.Uns);
|
||||
return (started.Host, nm, nm.TryGetVariable("eq-1/pv")!.NodeId);
|
||||
}
|
||||
|
||||
/// <summary>Pages until no continuation point or a Bad status; the page sizes and the final status.</summary>
|
||||
private static (List<double> Values, List<int> Pages, StatusCode Status) DrainFrom(
|
||||
OtOpcUaNodeManager nm, NodeId nodeId, DateTime start, DateTime end, uint max, byte[]? cp, int maxPages = 20,
|
||||
bool returnBounds = false)
|
||||
{
|
||||
var values = new List<double>();
|
||||
var pages = new List<int>();
|
||||
ServiceResult error;
|
||||
do
|
||||
{
|
||||
(var page, error, cp) = ReadRaw(nm, nodeId, start, end, max, cp, returnBounds);
|
||||
values.AddRange(page);
|
||||
pages.Add(page.Count);
|
||||
pages.Count.ShouldBeLessThanOrEqualTo(maxPages, "paging must terminate");
|
||||
}
|
||||
while (cp is not null && StatusCode.IsGood(error.StatusCode));
|
||||
return (values, pages, error.StatusCode);
|
||||
}
|
||||
|
||||
private static (IReadOnlyList<double> Values, ServiceResult Error, byte[]? Cp) ReadRaw(
|
||||
OtOpcUaNodeManager nm, NodeId nodeId, DateTime start, DateTime end, uint max, byte[]? inboundCp, bool returnBounds = false)
|
||||
{
|
||||
var context = new OperationContext(new RequestHeader(), secureChannelContext: null, RequestType.HistoryRead, identity: null);
|
||||
var details = new ReadRawModifiedDetails
|
||||
{
|
||||
ReturnBounds = returnBounds,
|
||||
IsReadModified = false,
|
||||
StartTime = start,
|
||||
EndTime = end,
|
||||
NumValuesPerNode = max,
|
||||
};
|
||||
var nodesToRead = new List<HistoryReadValueId> { new() { NodeId = nodeId, ContinuationPoint = inboundCp } };
|
||||
var results = new List<SdkHistoryReadResult> { null! };
|
||||
var errors = new List<ServiceResult> { null! };
|
||||
nm.HistoryRead(context, details, TimestampsToReturn.Both, releaseContinuationPoints: false, nodesToRead, results, errors);
|
||||
|
||||
var values = new List<double>();
|
||||
if (results[0]?.HistoryData is { } ho && !ExtensionObject.IsNull(ho))
|
||||
foreach (var dv in ((HistoryData)ExtensionObject.ToEncodeable(ho)).DataValues)
|
||||
values.Add(Convert.ToDouble(dv.Value));
|
||||
return (values, errors[0], results[0]?.ContinuationPoint);
|
||||
}
|
||||
|
||||
public void Dispose()
|
||||
{
|
||||
if (Directory.Exists(_pkiRoot))
|
||||
{
|
||||
try { Directory.Delete(_pkiRoot, recursive: true); }
|
||||
catch { /* best-effort */ }
|
||||
}
|
||||
}
|
||||
}
|
||||
+24
-5
@@ -222,17 +222,36 @@ public sealed class GatewayRawBoundaryRowTests
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task A_margin_still_crowding_the_re_read_fails_the_node_as_a_request_limit()
|
||||
public async Task A_margin_crowding_a_truncated_read_is_re_read_with_growing_room()
|
||||
{
|
||||
// 200 rows in the margin: the single bounded re-read (5 + 21 + 16) is still short of the cap because of them.
|
||||
// #694 F4: 200 rows in the margin. The first read (5 + 16) saw only 21 of them, so a single re-read sized from that
|
||||
// count (5 + 21 + 16) was still short and the node failed — with no ceiling at all. The room now at least doubles per
|
||||
// re-read (21, 42, 84, 168, 336), so the window is served in a bounded number of reads.
|
||||
var margin = Enumerable.Range(0, 200).Select(i => (At(30).AddTicks(-1 - i), (double)i, false));
|
||||
var fake = Aveva(margin.Append((At(35), 1d, false)).ToList());
|
||||
var ds = Source(fake);
|
||||
|
||||
await Should.ThrowAsync<HistoryReadLimitExceededException>(() => ds.ReadRawAsync("T", At(30), At(40), 5, Ct));
|
||||
var r = await ds.ReadRawAsync("T", At(30), At(40), 5, Ct);
|
||||
|
||||
fake.ReadRawCallCount.ShouldBe(2); // one bounded re-read, never more
|
||||
fake.RawStreamsCompleted.ShouldBe(2);
|
||||
r.Samples.Select(s => s.Value).ShouldBe([1d]);
|
||||
fake.ReadRawCallCount.ShouldBe(6);
|
||||
fake.RawStreamsCompleted.ShouldBe(6);
|
||||
ds.GetHealthSnapshot().TotalFailures.ShouldBe(0);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task A_margin_still_crowding_the_bounded_re_reads_fails_the_node_as_a_request_limit()
|
||||
{
|
||||
// 20,000 rows in the margin: past the bounded re-reads (21 · 2^8 ≈ 5,400 rows of room).
|
||||
var margin = Enumerable.Range(0, 20_000).Select(i => (At(30).AddTicks(-1 - i), (double)i, false));
|
||||
var fake = Aveva(margin.Append((At(35), 1d, false)).ToList());
|
||||
var ds = Source(fake);
|
||||
|
||||
var ex = await Should.ThrowAsync<HistoryReadLimitExceededException>(() => ds.ReadRawAsync("T", At(30), At(40), 5, Ct));
|
||||
|
||||
ex.PreWindowRows.ShouldBeGreaterThan(0);
|
||||
fake.ReadRawCallCount.ShouldBe(9); // the read + MaxCrowdedReReads, never more
|
||||
fake.RawStreamsCompleted.ShouldBe(9);
|
||||
ds.GetHealthSnapshot().TotalFailures.ShouldBe(0);
|
||||
}
|
||||
|
||||
|
||||
+5
@@ -31,4 +31,9 @@
|
||||
<ProjectReference Include="..\..\..\src\Drivers\ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway\ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.csproj"/>
|
||||
</ItemGroup>
|
||||
|
||||
<ItemGroup>
|
||||
<!-- #694: the node manager's HistoryRead paging over the REAL gateway data source boots a real SDK server. -->
|
||||
<Compile Include="..\..\Server\ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests\OpcUaHostStart.cs" Link="OpcUaHostStart.cs"/>
|
||||
</ItemGroup>
|
||||
|
||||
</Project>
|
||||
|
||||
+12
@@ -84,6 +84,18 @@ public sealed class OtOpcUaServerHostedServiceHistoryUpdateTests
|
||||
Service(view: null, config: config).HonorReturnBounds.ShouldBeFalse();
|
||||
}
|
||||
|
||||
/// <summary>#694: the events horizon is bound from <c>ServerHistorian:EventReadHorizon</c>, default none.</summary>
|
||||
[Fact]
|
||||
public void The_event_read_horizon_comes_from_options_and_defaults_off()
|
||||
{
|
||||
Service(view: null).EventReadHorizon.ShouldBe(TimeSpan.Zero);
|
||||
var config = new ConfigurationBuilder().AddInMemoryCollection(new Dictionary<string, string?>
|
||||
{
|
||||
["ServerHistorian:EventReadHorizon"] = "365.00:00:00",
|
||||
}).Build();
|
||||
Service(view: null, config: config).EventReadHorizon.ShouldBe(TimeSpan.FromDays(365));
|
||||
}
|
||||
|
||||
private static OtOpcUaServerHostedService Service(
|
||||
IHistoryUpdateRoleView? view, IDriverPrimaryRoleSource? primary = null, IConfiguration? config = null,
|
||||
TimeProvider? clock = null) => new(
|
||||
|
||||
@@ -0,0 +1,63 @@
|
||||
using Shouldly;
|
||||
using Xunit;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests;
|
||||
|
||||
/// <summary>Gitea #694: <see cref="EventHorizon.TryClamp"/> — the pure window arithmetic of <c>ServerHistorian:EventReadHorizon</c>.</summary>
|
||||
public sealed class EventHorizonTests
|
||||
{
|
||||
private static readonly DateTime Now = new(2026, 10, 13, 12, 0, 0, DateTimeKind.Utc);
|
||||
private static readonly TimeSpan H = TimeSpan.FromDays(30);
|
||||
private static readonly DateTime Floor = Now - H;
|
||||
|
||||
[Fact]
|
||||
public void No_horizon_changes_nothing()
|
||||
{
|
||||
var (s, e) = (DateTime.MinValue, Now);
|
||||
EventHorizon.TryClamp(ref s, ref e, reverse: false, TimeSpan.Zero, Now).ShouldBeTrue();
|
||||
(s, e).ShouldBe((DateTime.MinValue, Now));
|
||||
EventHorizon.TryClamp(ref s, ref e, reverse: false, TimeSpan.FromDays(-1), Now).ShouldBeTrue();
|
||||
(s, e).ShouldBe((DateTime.MinValue, Now));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void A_forward_window_starts_at_the_floor()
|
||||
{
|
||||
var (s, e) = (DateTime.UnixEpoch, Now);
|
||||
EventHorizon.TryClamp(ref s, ref e, reverse: false, H, Now).ShouldBeTrue();
|
||||
(s, e).ShouldBe((Floor, Now));
|
||||
|
||||
var (s2, e2) = (Floor.AddDays(1), Now);
|
||||
EventHorizon.TryClamp(ref s2, ref e2, reverse: false, H, Now).ShouldBeTrue();
|
||||
s2.ShouldBe(Floor.AddDays(1)); // already inside: unchanged
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void A_reverse_window_keeps_the_floor_instant()
|
||||
{
|
||||
var (s, e) = (Now, new DateTime(0, DateTimeKind.Utc));
|
||||
EventHorizon.TryClamp(ref s, ref e, reverse: true, H, Now).ShouldBeTrue();
|
||||
(s, e).ShouldBe((Now, Floor.AddTicks(-1))); // (end, start] — the floor itself is read
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void A_window_wholly_before_the_floor_is_empty()
|
||||
{
|
||||
var (s, e) = (Floor.AddDays(-2), Floor);
|
||||
EventHorizon.TryClamp(ref s, ref e, reverse: false, H, Now).ShouldBeFalse(); // [start, end) ends at the floor
|
||||
|
||||
var (rs, re) = (Floor.AddTicks(-1), Floor.AddDays(-2));
|
||||
EventHorizon.TryClamp(ref rs, ref re, reverse: true, H, Now).ShouldBeFalse();
|
||||
|
||||
var (ok, oe) = (Floor, Floor.AddDays(-2));
|
||||
EventHorizon.TryClamp(ref ok, ref oe, reverse: true, H, Now).ShouldBeTrue(); // (…, floor] holds the floor
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void A_horizon_longer_than_the_calendar_is_no_horizon()
|
||||
{
|
||||
var (s, e) = (DateTime.MinValue, Now);
|
||||
EventHorizon.TryClamp(ref s, ref e, reverse: false, TimeSpan.MaxValue, Now).ShouldBeTrue();
|
||||
s.ShouldBe(DateTime.MinValue);
|
||||
}
|
||||
}
|
||||
+231
@@ -0,0 +1,231 @@
|
||||
using Microsoft.Extensions.Logging.Abstractions;
|
||||
using Opc.Ua;
|
||||
using Opc.Ua.Server;
|
||||
using Shouldly;
|
||||
using Xunit;
|
||||
using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
|
||||
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;
|
||||
|
||||
/// <summary>
|
||||
/// Gitea #694 item 2 (live gate F4): a walked-back events read on the Gateway proves "nothing older" with one forward read
|
||||
/// down to the Unix epoch (#691) — ~3.8 s live on <c>Runtime.dbo.Events</c>, and an unfiltered deep read hit the gateway's
|
||||
/// 30 s SQL timeout. The opt-in <c>ServerHistorian:EventReadHorizon</c> (<see cref="OtOpcUaNodeManager.EventReadHorizon"/>)
|
||||
/// bounds every events read at <c>now − horizon</c>, in both directions; off (the default) nothing changes. The backend is
|
||||
/// shaped like the Gateway's SQL event read: forward only, <c>[start, end]</c> inclusive, refuses <c>start > end</c>
|
||||
/// (counted), records every requested window.
|
||||
/// </summary>
|
||||
public sealed class NodeManagerHistoryReadEventsHorizonTests : IDisposable
|
||||
{
|
||||
private static CancellationToken Ct => TestContext.Current.CancellationToken;
|
||||
private static readonly DateTime Now = new(2026, 10, 13, 12, 0, 0, DateTimeKind.Utc);
|
||||
private static readonly TimeSpan Horizon = TimeSpan.FromDays(30);
|
||||
private static readonly DateTime Floor = Now - Horizon;
|
||||
private readonly string _pkiRoot = Path.Combine(Path.GetTempPath(), $"otopcua-eventhorizon-{Guid.NewGuid():N}");
|
||||
|
||||
/// <summary>Five recent events (the last five minutes) — a source with little history, as in the live gate.</summary>
|
||||
private static List<HistoricalEvent> Recent() =>
|
||||
Enumerable.Range(0, 5).Select(i => Ev($"r{i}", Now.AddMinutes(-5 + i))).ToList();
|
||||
|
||||
[Fact]
|
||||
public async Task Without_a_horizon_a_walked_reverse_read_proves_the_rest_empty_down_to_the_epoch()
|
||||
{
|
||||
// The positive control: today's exact behaviour, and the cost the horizon removes.
|
||||
var (host, nm, node, gw) = await SetupAsync(Recent());
|
||||
try
|
||||
{
|
||||
var r = ReadEvents(nm, node, max: 10, start: DateTime.MinValue, end: Now);
|
||||
r.Error.StatusCode.Code.ShouldBe(StatusCodes.Good);
|
||||
r.Ids.ShouldBe(["r4", "r3", "r2", "r1", "r0"]);
|
||||
gw.Requests.Min(q => q.Start).ShouldBe(DateTime.UnixEpoch);
|
||||
gw.BadRequests.ShouldBe(0);
|
||||
}
|
||||
finally { await host.DisposeAsync(); }
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task With_a_horizon_the_walk_never_reads_before_it_and_answers_the_same()
|
||||
{
|
||||
var (host, nm, node, gw) = await SetupAsync(Recent());
|
||||
try
|
||||
{
|
||||
nm.EventReadHorizon = Horizon;
|
||||
var r = ReadEvents(nm, node, max: 10, start: DateTime.MinValue, end: Now);
|
||||
r.Error.StatusCode.Code.ShouldBe(StatusCodes.Good);
|
||||
r.Ids.ShouldBe(["r4", "r3", "r2", "r1", "r0"]);
|
||||
r.Cp.ShouldBeNull();
|
||||
// Requests are rounded outward to the millisecond, so the earliest one may start up to 1 ms before the floor.
|
||||
gw.Requests.Min(q => q.Start).ShouldBeGreaterThanOrEqualTo(Floor.AddMilliseconds(-1));
|
||||
gw.BadRequests.ShouldBe(0);
|
||||
}
|
||||
finally { await host.DisposeAsync(); }
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Events_older_than_the_horizon_are_not_returned_in_either_direction()
|
||||
{
|
||||
var events = new List<HistoricalEvent> { Ev("old", Floor.AddDays(-10)), Ev("edge", Floor), Ev("new", Now.AddDays(-1)) };
|
||||
var (host, nm, node, gw) = await SetupAsync(events);
|
||||
try
|
||||
{
|
||||
nm.EventReadHorizon = Horizon;
|
||||
|
||||
var reverse = ReadEvents(nm, node, max: 10, start: Now, end: DateTime.UnixEpoch);
|
||||
reverse.Ids.ShouldBe(["new", "edge"]); // the floor instant itself is inside
|
||||
|
||||
var forward = ReadEvents(nm, node, max: 10, start: DateTime.UnixEpoch, end: Now);
|
||||
forward.Ids.ShouldBe(["edge", "new"]);
|
||||
|
||||
gw.Requests.ShouldAllBe(q => q.Start >= Floor.AddMilliseconds(-1));
|
||||
|
||||
nm.EventReadHorizon = TimeSpan.Zero; // off again: exact to the window
|
||||
ReadEvents(nm, node, max: 10, start: DateTime.UnixEpoch, end: Now).Ids.ShouldBe(["old", "edge", "new"]);
|
||||
}
|
||||
finally { await host.DisposeAsync(); }
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task A_window_wholly_older_than_the_horizon_is_GoodNoData_and_never_sent()
|
||||
{
|
||||
var (host, nm, node, gw) = await SetupAsync([Ev("old", Floor.AddDays(-10))]);
|
||||
try
|
||||
{
|
||||
nm.EventReadHorizon = Horizon;
|
||||
var forward = ReadEvents(nm, node, max: 10, start: Floor.AddDays(-20), end: Floor.AddDays(-1));
|
||||
forward.Error.StatusCode.Code.ShouldBe(StatusCodes.Good);
|
||||
forward.Status.Code.ShouldBe(StatusCodes.GoodNoData);
|
||||
forward.Ids.ShouldBeEmpty();
|
||||
|
||||
var reverse = ReadEvents(nm, node, max: 10, start: Floor.AddDays(-1), end: Floor.AddDays(-20));
|
||||
reverse.Status.Code.ShouldBe(StatusCodes.GoodNoData);
|
||||
|
||||
gw.Requests.ShouldBeEmpty();
|
||||
}
|
||||
finally { await host.DisposeAsync(); }
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task A_page_through_under_a_horizon_keeps_its_floor_across_continuation_points()
|
||||
{
|
||||
var events = Enumerable.Range(0, 25).Select(i => Ev($"e{i:00}", Now.AddMinutes(-i))).ToList();
|
||||
events.Add(Ev("old", Floor.AddDays(-1)));
|
||||
var (host, nm, node, gw) = await SetupAsync(events);
|
||||
try
|
||||
{
|
||||
nm.EventReadHorizon = Horizon;
|
||||
var all = new List<string>();
|
||||
byte[]? cp = null;
|
||||
var pages = 0;
|
||||
do
|
||||
{
|
||||
var r = ReadEvents(nm, node, max: 10, start: DateTime.MinValue, end: Now, inboundCp: cp);
|
||||
r.Error.StatusCode.Code.ShouldBe(StatusCodes.Good);
|
||||
all.AddRange(r.Ids);
|
||||
cp = r.Cp;
|
||||
(++pages).ShouldBeLessThan(10);
|
||||
}
|
||||
while (cp is not null);
|
||||
|
||||
all.ShouldBe(Enumerable.Range(0, 25).Select(i => $"e{i:00}"));
|
||||
gw.Requests.ShouldAllBe(q => q.Start >= Floor.AddMilliseconds(-1));
|
||||
}
|
||||
finally { await host.DisposeAsync(); }
|
||||
}
|
||||
|
||||
// ---- helpers ---------------------------------------------------------------------------------------------------------
|
||||
|
||||
private static HistoricalEvent Ev(string id, DateTime at) => new(id, "src", at, at, "m", 500);
|
||||
|
||||
private async Task<(OpcUaApplicationHost Host, OtOpcUaNodeManager Nm, NodeId Node, GatewayShapedEventSource Gw)> SetupAsync(
|
||||
List<HistoricalEvent> events)
|
||||
{
|
||||
var started = await OpcUaHostStart.StartAsync(
|
||||
port => new OpcUaApplicationHost(
|
||||
new OpcUaApplicationHostOptions
|
||||
{
|
||||
ApplicationName = "OtOpcUa.EventHorizonTest",
|
||||
ApplicationUri = $"urn:OtOpcUa.EventHorizonTest:{Guid.NewGuid():N}",
|
||||
OpcUaPort = port,
|
||||
PublicHostname = "localhost",
|
||||
PkiStoreRoot = _pkiRoot,
|
||||
},
|
||||
NullLogger<OpcUaApplicationHost>.Instance),
|
||||
() => new OtOpcUaSdkServer(), Ct);
|
||||
var nm = started.Server.NodeManager!;
|
||||
nm.HistoryContinuationStore = new InMemoryHistoryContinuationStore();
|
||||
nm.Clock = new FixedClock(Now);
|
||||
var gw = new GatewayShapedEventSource(events);
|
||||
nm.HistorianDataSource = gw;
|
||||
nm.EnsureFolder("eq-evt", parentNodeId: null, displayName: "Equipment", realm: AddressSpaceRealm.Uns);
|
||||
nm.MaterialiseAlarmCondition("alarm-1", "eq-evt", "HighTemp", "OffNormalAlarm", severity: 700, realm: AddressSpaceRealm.Uns);
|
||||
return (started.Host, nm, nm.TryGetFolder("eq-evt")!.NodeId, gw);
|
||||
}
|
||||
|
||||
private static (List<string> Ids, ServiceResult Error, StatusCode Status, byte[]? Cp) ReadEvents(
|
||||
OtOpcUaNodeManager nm, NodeId node, uint max, DateTime start, DateTime end, byte[]? inboundCp = null)
|
||||
{
|
||||
var context = new OperationContext(new RequestHeader(), secureChannelContext: null, RequestType.HistoryRead, identity: null);
|
||||
var filter = new EventFilter();
|
||||
filter.SelectClauses.Add(new SimpleAttributeOperand(ObjectTypeIds.BaseEventType, new QualifiedName("EventId"))
|
||||
{
|
||||
AttributeId = Attributes.Value,
|
||||
});
|
||||
var details = new ReadEventDetails { StartTime = start, EndTime = end, NumValuesPerNode = max, Filter = filter };
|
||||
var nodesToRead = new List<HistoryReadValueId> { new() { NodeId = node, ContinuationPoint = inboundCp } };
|
||||
var results = new List<SdkHistoryReadResult> { null! };
|
||||
var errors = new List<ServiceResult> { null! };
|
||||
nm.HistoryRead(context, details, TimestampsToReturn.Both, false, nodesToRead, results, errors);
|
||||
|
||||
var ids = new List<string>();
|
||||
if (results[0]?.HistoryData is { } ho && !ExtensionObject.IsNull(ho))
|
||||
foreach (var e in ((HistoryEvent)ExtensionObject.ToEncodeable(ho)).Events)
|
||||
ids.Add(System.Text.Encoding.UTF8.GetString((byte[])e.EventFields[0].Value));
|
||||
return (ids, errors[0], results[0]?.StatusCode ?? StatusCodes.Good, results[0]?.ContinuationPoint);
|
||||
}
|
||||
|
||||
/// <summary>The Gateway's SQL event read: forward only, <c>[start, end]</c> inclusive, time-ordered, capped + flag.</summary>
|
||||
private sealed class GatewayShapedEventSource(List<HistoricalEvent> events) : IHistorianDataSource
|
||||
{
|
||||
public List<(DateTime Start, DateTime End, int Max)> Requests { get; } = [];
|
||||
public int BadRequests { get; private set; }
|
||||
public bool SupportsReverseEvents => false;
|
||||
|
||||
public Task<HistoricalEventsResult> ReadEventsAsync(string? n, DateTime s, DateTime e, int m, CancellationToken ct)
|
||||
{
|
||||
Requests.Add((s, e, m));
|
||||
if (s > e) BadRequests++; // the gateway's ValidateRange
|
||||
var window = events.Where(x => x.EventTimeUtc >= s && x.EventTimeUtc <= e).OrderBy(x => x.EventTimeUtc).ToList();
|
||||
var cap = m <= 0 ? int.MaxValue : m;
|
||||
return Task.FromResult(new HistoricalEventsResult(window.Take(cap).ToList(), window.Count > cap ? new byte[] { 1 } : null));
|
||||
}
|
||||
|
||||
public Task<HistoricalEventsResult> ReadEventsAsync(string? n, DateTime s, DateTime e, int m, bool reverse, CancellationToken ct) =>
|
||||
reverse ? throw new NotSupportedException("forward only") : ReadEventsAsync(n, s, e, m, ct);
|
||||
|
||||
public Task<HistorianRead> ReadRawAsync(string r, DateTime s, DateTime e, uint m, CancellationToken ct) =>
|
||||
NullHistorianDataSource.Instance.ReadRawAsync(r, s, e, m, ct);
|
||||
public Task<HistorianRead> ReadProcessedAsync(string r, DateTime s, DateTime e, TimeSpan i, HistoryAggregateType a, CancellationToken ct) =>
|
||||
NullHistorianDataSource.Instance.ReadProcessedAsync(r, s, e, i, a, ct);
|
||||
public Task<HistorianRead> ReadAtTimeAsync(string r, IReadOnlyList<DateTime> t, CancellationToken ct) =>
|
||||
NullHistorianDataSource.Instance.ReadAtTimeAsync(r, t, ct);
|
||||
public HistorianHealthSnapshot GetHealthSnapshot() => NullHistorianDataSource.Instance.GetHealthSnapshot();
|
||||
public void Dispose() { }
|
||||
}
|
||||
|
||||
private sealed class FixedClock(DateTime now) : TimeProvider
|
||||
{
|
||||
public override DateTimeOffset GetUtcNow() => new(now);
|
||||
}
|
||||
|
||||
public void Dispose()
|
||||
{
|
||||
if (Directory.Exists(_pkiRoot))
|
||||
{
|
||||
try { Directory.Delete(_pkiRoot, recursive: true); }
|
||||
catch { /* best-effort */ }
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,142 @@
|
||||
using Shouldly;
|
||||
using Xunit;
|
||||
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests;
|
||||
|
||||
/// <summary>
|
||||
/// Gitea #694: <see cref="RawPageLimit"/> — the page plan (P = min(want, L − k − 1), request = P + k + 1 ≤ L) and the one
|
||||
/// re-plan after a refusal. The end-to-end behaviour over the real Gateway data source is pinned in
|
||||
/// <c>GatewayCeilingPagingTests</c>.
|
||||
/// </summary>
|
||||
public sealed class RawPageLimitTests
|
||||
{
|
||||
[Theory]
|
||||
// want, k, L ⇒ page, request, noRoom
|
||||
[InlineData(5000u, 0, 0, 5000u, 5001u, false)] // no limit: #692 unchanged (page + ties + lookahead)
|
||||
[InlineData(5000u, 3, 0, 5000u, 5004u, false)]
|
||||
[InlineData(0u, 3, 0, 0u, 0u, false)] // no limit, uncapped: one uncapped read
|
||||
[InlineData(5000u, 0, 984, 983u, 984u, false)] // over the limit: the request fills the limit exactly
|
||||
[InlineData(5000u, 10, 984, 973u, 984u, false)] // boundary ties take room from the page
|
||||
[InlineData(10u, 10, 984, 10u, 21u, false)] // under the limit: unchanged
|
||||
[InlineData(0u, 0, 984, 983u, 984u, false)] // uncapped: a limit-sized page
|
||||
[InlineData(0u, 982, 984, 1u, 984u, false)]
|
||||
[InlineData(5000u, 983, 984, 0u, 0u, true)] // the ties alone fill a read: never a 0 (uncapped) request
|
||||
[InlineData(5000u, 5000, 984, 0u, 0u, true)] // never an underflow
|
||||
[InlineData(5000u, 0, 1, 0u, 0u, true)]
|
||||
public void Plan_fits_page_ties_and_lookahead_into_the_limit(uint want, int k, int limit, uint page, uint request, bool noRoom)
|
||||
{
|
||||
var plan = RawPageLimit.Plan(want, k, limit);
|
||||
plan.PageCap.ShouldBe(page);
|
||||
plan.RequestCap.ShouldBe(request);
|
||||
plan.NoRoom.ShouldBe(noRoom);
|
||||
if (limit > 0 && !noRoom) plan.RequestCap.ShouldBeLessThanOrEqualTo((uint)limit);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task A_read_that_teaches_the_limit_is_re_planned_once_at_it()
|
||||
{
|
||||
var source = new LimitSource();
|
||||
var asked = new List<uint>();
|
||||
var (page, plan) = await RawPageLimit.ReadAsync(source, 5000u, 0, p =>
|
||||
{
|
||||
asked.Add(p.RequestCap);
|
||||
if (p.RequestCap > 984) { source.Limit = 984; throw new HistoryReadLimitExceededException(1000, "ceiling"); }
|
||||
return Task.FromResult("ok");
|
||||
});
|
||||
page.ShouldBe("ok");
|
||||
plan.PageCap.ShouldBe(983u);
|
||||
asked.ShouldBe([5001u, 984u]);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task A_crowded_refusal_shrinks_the_page_by_the_pre_window_rows()
|
||||
{
|
||||
var source = new LimitSource { Limit = 984 };
|
||||
var asked = new List<uint>();
|
||||
var (_, plan) = await RawPageLimit.ReadAsync(source, 0u, 2, p =>
|
||||
{
|
||||
asked.Add(p.RequestCap);
|
||||
if (asked.Count == 1) throw new HistoryReadLimitExceededException(1000, "crowded") { PreWindowRows = 21 };
|
||||
return Task.FromResult("ok");
|
||||
});
|
||||
asked.ShouldBe([984u, 963u]);
|
||||
plan.PageCap.ShouldBe(960u);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task A_refusal_at_an_unchanged_limit_without_crowding_is_the_answer()
|
||||
{
|
||||
var source = new LimitSource { Limit = 984 };
|
||||
var calls = 0;
|
||||
await Should.ThrowAsync<HistoryReadLimitExceededException>(() => RawPageLimit.ReadAsync<string>(source, 5000u, 0, _ =>
|
||||
{
|
||||
calls++;
|
||||
throw new HistoryReadLimitExceededException(1000, "tie cluster");
|
||||
}));
|
||||
calls.ShouldBe(1);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task A_second_refusal_is_not_retried_again()
|
||||
{
|
||||
var source = new LimitSource();
|
||||
var calls = 0;
|
||||
await Should.ThrowAsync<HistoryReadLimitExceededException>(() => RawPageLimit.ReadAsync<string>(source, 0u, 0, _ =>
|
||||
{
|
||||
calls++;
|
||||
source.Limit = 984 - calls; // even if the limit keeps moving
|
||||
throw new HistoryReadLimitExceededException(1000, "refused") { PreWindowRows = 5 };
|
||||
}));
|
||||
calls.ShouldBe(2);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task No_room_is_returned_without_a_read()
|
||||
{
|
||||
var source = new LimitSource { Limit = 10 };
|
||||
var calls = 0;
|
||||
var (page, plan) = await RawPageLimit.ReadAsync(source, 5u, 9, _ => { calls++; return Task.FromResult("x"); });
|
||||
plan.NoRoom.ShouldBeTrue();
|
||||
page.ShouldBeNull();
|
||||
calls.ShouldBe(0);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Other_failures_are_not_retried()
|
||||
{
|
||||
var source = new LimitSource();
|
||||
var calls = 0;
|
||||
await Should.ThrowAsync<InvalidOperationException>(() => RawPageLimit.ReadAsync<string>(source, 5000u, 0, _ =>
|
||||
{
|
||||
calls++;
|
||||
source.Limit = 984;
|
||||
throw new InvalidOperationException("backend fault");
|
||||
}));
|
||||
calls.ShouldBe(1);
|
||||
}
|
||||
|
||||
/// <summary>Only <see cref="IHistorianDataSource.MaxRawValuesPerRead"/> matters here.</summary>
|
||||
private sealed class LimitSource : IHistorianDataSource
|
||||
{
|
||||
public int Limit;
|
||||
|
||||
public int MaxRawValuesPerRead => Limit;
|
||||
|
||||
public Task<HistoryReadResult> ReadRawAsync(string fullReference, DateTime startUtc, DateTime endUtc, uint maxValuesPerNode,
|
||||
CancellationToken cancellationToken) => throw new NotSupportedException();
|
||||
|
||||
public Task<HistoryReadResult> ReadProcessedAsync(string fullReference, DateTime startUtc, DateTime endUtc, TimeSpan interval,
|
||||
HistoryAggregateType aggregate, CancellationToken cancellationToken) => throw new NotSupportedException();
|
||||
|
||||
public Task<HistoryReadResult> 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() { }
|
||||
}
|
||||
}
|
||||
@@ -119,6 +119,24 @@ public sealed class ServerHistorianOptionsTests
|
||||
{ Enabled = true, Endpoint = "https://h:5222", ApiKey = "histgw_x_y", HonorReturnBounds = false }.Validate());
|
||||
}
|
||||
|
||||
/// <summary>#694: the events horizon defaults off, binds as a TimeSpan, and only a negative value warns.</summary>
|
||||
[Fact]
|
||||
public void EventReadHorizon_defaults_off_binds_and_only_a_negative_value_warns()
|
||||
{
|
||||
Assert.Equal(TimeSpan.Zero, new ServerHistorianOptions().EventReadHorizon);
|
||||
var bound = new ConfigurationBuilder()
|
||||
.AddInMemoryCollection(new Dictionary<string, string?> { ["ServerHistorian:EventReadHorizon"] = "3650.00:00:00" })
|
||||
.Build()
|
||||
.GetSection(ServerHistorianOptions.SectionName)
|
||||
.Get<ServerHistorianOptions>();
|
||||
Assert.Equal(TimeSpan.FromDays(3650), bound!.EventReadHorizon);
|
||||
Assert.Empty(new ServerHistorianOptions
|
||||
{ Enabled = true, Endpoint = "https://h:5222", ApiKey = "histgw_x_y", EventReadHorizon = TimeSpan.FromDays(30) }.Validate());
|
||||
var w = new ServerHistorianOptions
|
||||
{ Enabled = true, Endpoint = "https://h:5222", ApiKey = "histgw_x_y", EventReadHorizon = TimeSpan.FromDays(-1) }.Validate();
|
||||
Assert.Contains(w, m => m.Contains("EventReadHorizon"));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void HistoryUpdateMaxFutureSkew_defaults_to_one_day_binds_and_zero_is_silent()
|
||||
{
|
||||
|
||||
Reference in New Issue
Block a user