Joseph Doherty
d277f34a96
A reverse or end-only capped events read on a backend without SupportsReverseEvents (Gateway, Null) was refused BadHistoryOperationUnsupported (explicit reverse) or answered with the OLDEST N ascending (end-only, #675). It is now served newest-first by the same ReverseWalk as raw reads: page-sized ms-aligned segments, never answered from a read the source flagged truncated, ties sorted by EventId (the gateway orders by event time only - the fake shuffles them per call), one extra event walked to signal 'more', uncapped = one forward read reversed, and the width + proven-empty stretch carried in the events cursor. The oversized-cluster over-fetch keeps exactly the events at T in the same order. A tie cluster over the budget is BadTooManyOperations. Docs note the gateway's EventReadMaxRows must be unlimited (it truncates silently).
760 lines
38 KiB
C#
760 lines
38 KiB
C#
using Microsoft.Extensions.Logging;
|
|
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 #669 — server-side continuation-point paging for HistoryRead-Events. Same harness as
|
|
/// <see cref="NodeManagerHistoryReadPagingTests"/>: a real SDK server, a session-less in-process
|
|
/// HistoryRead and an <see cref="InMemoryHistoryContinuationStore"/>, over a fake source that behaves like
|
|
/// the SQL backends (ordered by (time, id), half-open window, cap + truncation flag).
|
|
/// </summary>
|
|
public sealed class NodeManagerHistoryReadEventsPagingTests : IDisposable
|
|
{
|
|
private static CancellationToken Ct => TestContext.Current.CancellationToken;
|
|
private static readonly DateTime Epoch = new(2026, 1, 1, 0, 0, 0, DateTimeKind.Utc);
|
|
private readonly string _pkiRoot = Path.Combine(Path.GetTempPath(), $"otopcua-eventpaging-{Guid.NewGuid():N}");
|
|
|
|
[Fact]
|
|
public async Task Events_page_across_continuation_points_exactly_once_in_order()
|
|
{
|
|
var (host, nm, node, src) = await SetupAsync(Distinct(250));
|
|
var ids = DrainAsync(nm, node, max: 100);
|
|
ids.ShouldBe(src.All.Select(e => e.EventId));
|
|
src.Calls.ShouldBeGreaterThan(2);
|
|
await host.DisposeAsync();
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Events_tie_at_page_boundary_is_neither_duplicated_nor_skipped()
|
|
{
|
|
// Indices 2,3,4 share a timestamp; cap 4 cuts through the cluster.
|
|
var events = new List<HistoricalEvent>
|
|
{
|
|
Ev("e0", 0), Ev("e1", 1), Ev("e2", 2), Ev("e3", 2), Ev("e4", 2), Ev("e5", 3),
|
|
};
|
|
var (host, nm, node, _) = await SetupAsync(events);
|
|
DrainAsync(nm, node, max: 4).ShouldBe(["e0", "e1", "e2", "e3", "e4", "e5"]);
|
|
await host.DisposeAsync();
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Events_tie_cluster_larger_than_the_cap_pages_within_the_cluster()
|
|
{
|
|
var events = new List<HistoricalEvent> { Ev("a0", 0) };
|
|
for (var i = 0; i < 7; i++) events.Add(Ev($"t{i}", 5));
|
|
events.Add(Ev("z0", 9));
|
|
var (host, nm, node, _) = await SetupAsync(events);
|
|
DrainAsync(nm, node, max: 3).ShouldBe(
|
|
["a0", "t0", "t1", "t2", "t3", "t4", "t5", "t6", "z0"]);
|
|
await host.DisposeAsync();
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Events_tie_cluster_beyond_the_overfetch_bound_fails_loudly()
|
|
{
|
|
var events = new List<HistoricalEvent>();
|
|
for (var i = 0; i < 10; i++) events.Add(Ev($"t{i:00}", 5));
|
|
var (host, nm, node, _) = await SetupAsync(events);
|
|
nm.MaxTieClusterOverfetch = 4;
|
|
|
|
var p1 = ReadEvents(nm, node, max: 3, inboundCp: null);
|
|
p1.Cp.ShouldNotBeNull();
|
|
var p2 = ReadEvents(nm, node, max: 3, inboundCp: p1.Cp);
|
|
p2.Error.StatusCode.Code.ShouldBe(StatusCodes.BadTooManyOperations);
|
|
await host.DisposeAsync();
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Events_unknown_continuation_point_is_BadContinuationPointInvalid()
|
|
{
|
|
var (host, nm, node, _) = await SetupAsync(Distinct(5));
|
|
var r = ReadEvents(nm, node, max: 2, inboundCp: Guid.NewGuid().ToByteArray());
|
|
r.Error.StatusCode.Code.ShouldBe(StatusCodes.BadContinuationPointInvalid);
|
|
await host.DisposeAsync();
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Events_release_drops_the_cursor()
|
|
{
|
|
var (host, nm, node, _) = await SetupAsync(Distinct(10));
|
|
var p1 = ReadEvents(nm, node, max: 4, inboundCp: null);
|
|
p1.Cp.ShouldNotBeNull();
|
|
InvokeRelease(nm, node, p1.Cp!);
|
|
ReadEvents(nm, node, max: 4, inboundCp: p1.Cp).Error.StatusCode.Code
|
|
.ShouldBe(StatusCodes.BadContinuationPointInvalid);
|
|
await host.DisposeAsync();
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Events_cap_sized_complete_page_emits_no_continuation_point()
|
|
{
|
|
var (host, nm, node, _) = await SetupAsync(Distinct(4));
|
|
var r = ReadEvents(nm, node, max: 4, inboundCp: null);
|
|
r.Ids.Count.ShouldBe(4);
|
|
r.Cp.ShouldBeNull("the source did not flag truncation, so a cap-sized page is not a paged one");
|
|
await host.DisposeAsync();
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Events_unbounded_request_never_pages()
|
|
{
|
|
var (host, nm, node, _) = await SetupAsync(Distinct(300));
|
|
var r = ReadEvents(nm, node, max: 0, inboundCp: null);
|
|
r.Ids.Count.ShouldBe(300);
|
|
r.Cp.ShouldBeNull();
|
|
await host.DisposeAsync();
|
|
}
|
|
|
|
[Fact]
|
|
public async Task A_raw_cursor_presented_to_an_events_read_is_invalid()
|
|
{
|
|
var (host, nm, node, _) = await SetupAsync(Distinct(10));
|
|
var raw = nm.HistoryContinuationStore.Save(null, new HistoryContinuationState("t", Epoch, Epoch.AddDays(1), 0, 4));
|
|
ReadEvents(nm, node, max: 4, inboundCp: raw).Error.StatusCode.Code
|
|
.ShouldBe(StatusCodes.BadContinuationPointInvalid);
|
|
await host.DisposeAsync();
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Events_open_ended_forward_read_pages_without_duplicates()
|
|
{
|
|
// EndTime unset (MinValue) + a StartTime = "page forward from StartTime" (Part 11).
|
|
var (host, nm, node, src) = await SetupAsync(Distinct(25));
|
|
DrainAsync(nm, node, max: 10, end: DateTime.MinValue).ShouldBe(src.All.Select(e => e.EventId));
|
|
await host.DisposeAsync();
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Events_cursor_is_bound_to_its_node()
|
|
{
|
|
var (host, nm, node, _) = await SetupAsync(Distinct(10));
|
|
var other = SecondNode(nm);
|
|
var p1 = ReadEvents(nm, node, max: 4, inboundCp: null);
|
|
p1.Cp.ShouldNotBeNull();
|
|
ReadEvents(nm, other, max: 4, inboundCp: p1.Cp).Error.StatusCode.Code
|
|
.ShouldBe(StatusCodes.BadContinuationPointInvalid);
|
|
await host.DisposeAsync();
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Raw_cursor_is_bound_to_its_node()
|
|
{
|
|
var (host, nm, _, _) = await SetupAsync(Distinct(1));
|
|
nm.HistorianDataSource = new RawSeriesSource();
|
|
nm.EnsureVariable("eq-1/a", null, "A", "Double", false, historianTagname: "WW.A", realm: AddressSpaceRealm.Uns);
|
|
nm.EnsureVariable("eq-1/b", null, "B", "Double", false, historianTagname: "WW.B", realm: AddressSpaceRealm.Uns);
|
|
var a = nm.TryGetVariable("eq-1/a")!.NodeId;
|
|
var b = nm.TryGetVariable("eq-1/b")!.NodeId;
|
|
|
|
byte[] RawRead(NodeId n, byte[]? cp, out ServiceResult err)
|
|
{
|
|
var ctx = new OperationContext(new RequestHeader(), secureChannelContext: null, RequestType.HistoryRead, identity: null);
|
|
var res = new List<SdkHistoryReadResult> { null! };
|
|
var errs = new List<ServiceResult> { null! };
|
|
nm.HistoryRead(ctx, new ReadRawModifiedDetails { StartTime = Epoch, EndTime = Epoch.AddHours(1), NumValuesPerNode = 2, IsReadModified = false },
|
|
TimestampsToReturn.Both, false,
|
|
new List<HistoryReadValueId> { new() { NodeId = n, ContinuationPoint = cp } }, res, errs);
|
|
err = errs[0];
|
|
return res[0]?.ContinuationPoint!;
|
|
}
|
|
|
|
var cp = RawRead(a, null, out var err0);
|
|
err0.StatusCode.Code.ShouldBe(StatusCodes.Good, err0.ToString());
|
|
cp.ShouldNotBeNull();
|
|
RawRead(b, cp, out var err);
|
|
err.StatusCode.Code.ShouldBe(StatusCodes.BadContinuationPointInvalid);
|
|
await host.DisposeAsync();
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Events_tie_cluster_pages_on_a_coarse_resolution_backend()
|
|
{
|
|
const long res = 10; // QuestDB-like: inputs truncate to 10 ticks
|
|
var events = new List<HistoricalEvent>();
|
|
DateTime At(int slot) => Epoch.AddTicks(slot * res);
|
|
events.Add(new("a0", "s", At(0), Epoch, "m", 500));
|
|
for (var i = 0; i < 7; i++) events.Add(new($"t{i}", "s", At(3), Epoch, "m", 500));
|
|
events.Add(new("z0", "s", At(8), Epoch, "m", 500));
|
|
var (host, nm, node, _) = await SetupAsync(events, res);
|
|
DrainAsync(nm, node, max: 3).ShouldBe(["a0", "t0", "t1", "t2", "t3", "t4", "t5", "t6", "z0"]);
|
|
await host.DisposeAsync();
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Events_batch_one_node_failing_does_not_disturb_another_that_pages()
|
|
{
|
|
var (host, nm, node, src) = await SetupAsync(Distinct(10));
|
|
var other = SecondNode(nm);
|
|
var good1 = ReadBatch(nm, 4, null, (node, null), (other, null));
|
|
good1[0].Cp.ShouldNotBeNull();
|
|
good1[1].Cp.ShouldNotBeNull();
|
|
|
|
var mixed = ReadBatch(nm, 4, null, (node, good1[0].Cp), (other, Guid.NewGuid().ToByteArray()));
|
|
mixed[0].Error.StatusCode.Code.ShouldBe(StatusCodes.Good);
|
|
// A resumed page is FULL (#675): it over-fetches by the boundary skip it then trims.
|
|
mixed[0].Ids.ShouldBe(src.All.Skip(4).Take(4).Select(e => e.EventId));
|
|
mixed[1].Error.StatusCode.Code.ShouldBe(StatusCodes.BadContinuationPointInvalid);
|
|
await host.DisposeAsync();
|
|
}
|
|
|
|
// --- Gitea #675: reverse (newest-first) event paging ------------------------------------------------------
|
|
|
|
[Fact]
|
|
public async Task Reverse_events_page_newest_first_exactly_once_with_exact_page_counts()
|
|
{
|
|
var (host, nm, node, src) = await SetupAsync(Distinct(250));
|
|
var pages = new List<int>();
|
|
var ids = DrainAsync(nm, node, max: 100, start: Late, end: Early, pageSizes: pages);
|
|
ids.ShouldBe(src.All.Reverse().Select(e => e.EventId));
|
|
pages.ShouldBe([100, 100, 50]);
|
|
await host.DisposeAsync();
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Reverse_first_page_is_the_LATEST_events_with_a_continuation_point()
|
|
{
|
|
// The pre-#675 shape: the cap kept the EARLIEST N, ascending, Good, and no continuation point.
|
|
var (host, nm, node, _) = await SetupAsync(Distinct(10));
|
|
var p1 = ReadEvents(nm, node, max: 3, inboundCp: null, end: Early, start: Late);
|
|
p1.Error.StatusCode.Code.ShouldBe(StatusCodes.Good);
|
|
p1.Ids.ShouldBe(["e0009", "e0008", "e0007"]);
|
|
p1.Cp.ShouldNotBeNull("a capped reverse read has more behind it and must page");
|
|
await host.DisposeAsync();
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Reverse_tie_at_page_boundary_is_neither_duplicated_nor_skipped()
|
|
{
|
|
var events = new List<HistoricalEvent>
|
|
{
|
|
Ev("e0", 0), Ev("e1", 1), Ev("e2", 2), Ev("e3", 2), Ev("e4", 2), Ev("e5", 3),
|
|
};
|
|
var (host, nm, node, _) = await SetupAsync(events);
|
|
// Newest-first, ties in descending EventId: cap 2 cuts the cluster at e4|e3.
|
|
DrainAsync(nm, node, max: 2, start: Late, end: Early).ShouldBe(["e5", "e4", "e3", "e2", "e1", "e0"]);
|
|
await host.DisposeAsync();
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Reverse_tie_cluster_larger_than_the_cap_pages_within_the_cluster()
|
|
{
|
|
var events = new List<HistoricalEvent> { Ev("a0", 0) };
|
|
for (var i = 0; i < 7; i++) events.Add(Ev($"t{i}", 5));
|
|
events.Add(Ev("z0", 9));
|
|
var (host, nm, node, _) = await SetupAsync(events);
|
|
var pages = new List<int>();
|
|
DrainAsync(nm, node, max: 3, start: Late, end: Early, pageSizes: pages).ShouldBe(
|
|
["z0", "t6", "t5", "t4", "t3", "t2", "t1", "t0", "a0"]);
|
|
pages.Sum().ShouldBe(9);
|
|
await host.DisposeAsync();
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Reverse_tie_cluster_pages_on_a_coarse_resolution_backend()
|
|
{
|
|
const long res = 10;
|
|
var events = new List<HistoricalEvent>();
|
|
DateTime At(int slot) => Epoch.AddTicks(slot * res);
|
|
events.Add(new("a0", "s", At(0), Epoch, "m", 500));
|
|
for (var i = 0; i < 7; i++) events.Add(new($"t{i}", "s", At(3), Epoch, "m", 500));
|
|
events.Add(new("z0", "s", At(8), Epoch, "m", 500));
|
|
var (host, nm, node, _) = await SetupAsync(events, res);
|
|
DrainAsync(nm, node, max: 3, start: Late, end: Early).ShouldBe(
|
|
["z0", "t6", "t5", "t4", "t3", "t2", "t1", "t0", "a0"]);
|
|
await host.DisposeAsync();
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Reverse_tie_cluster_beyond_the_overfetch_bound_fails_loudly()
|
|
{
|
|
var events = new List<HistoricalEvent>();
|
|
for (var i = 0; i < 10; i++) events.Add(Ev($"t{i:00}", 5));
|
|
var (host, nm, node, _) = await SetupAsync(events);
|
|
nm.MaxTieClusterOverfetch = 4;
|
|
var p1 = ReadEvents(nm, node, max: 3, inboundCp: null, end: Early, start: Late);
|
|
p1.Cp.ShouldNotBeNull();
|
|
ReadEvents(nm, node, max: 3, inboundCp: p1.Cp, end: Early, start: Late).Error.StatusCode.Code
|
|
.ShouldBe(StatusCodes.BadTooManyOperations);
|
|
await host.DisposeAsync();
|
|
}
|
|
|
|
[Fact]
|
|
public async Task A_reverse_cursor_resumes_reverse_whatever_the_resume_request_says()
|
|
{
|
|
// The direction lives in the cursor: a resume carrying FORWARD details still continues newest-first.
|
|
var (host, nm, node, _) = await SetupAsync(Distinct(10));
|
|
var p1 = ReadEvents(nm, node, max: 4, inboundCp: null, end: Early, start: Late);
|
|
p1.Ids.ShouldBe(["e0009", "e0008", "e0007", "e0006"]);
|
|
var p2 = ReadEvents(nm, node, max: 4, inboundCp: p1.Cp); // default details: Early -> Late (forward)
|
|
p2.Error.StatusCode.Code.ShouldBe(StatusCodes.Good);
|
|
p2.Ids.ShouldBe(["e0005", "e0004", "e0003", "e0002"]);
|
|
await host.DisposeAsync();
|
|
}
|
|
|
|
[Fact]
|
|
public async Task A_forward_cursor_resumes_forward_whatever_the_resume_request_says()
|
|
{
|
|
var (host, nm, node, _) = await SetupAsync(Distinct(10));
|
|
var p1 = ReadEvents(nm, node, max: 4, inboundCp: null);
|
|
p1.Ids.ShouldBe(["e0000", "e0001", "e0002", "e0003"]);
|
|
var p2 = ReadEvents(nm, node, max: 4, inboundCp: p1.Cp, end: Early, start: Late);
|
|
p2.Ids.ShouldBe(["e0004", "e0005", "e0006", "e0007"]);
|
|
await host.DisposeAsync();
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Reverse_unbounded_read_returns_everything_newest_first_without_paging()
|
|
{
|
|
var (host, nm, node, src) = await SetupAsync(Distinct(30));
|
|
var r = ReadEvents(nm, node, max: 0, inboundCp: null, end: Early, start: Late);
|
|
r.Ids.ShouldBe(src.All.Reverse().Select(e => e.EventId));
|
|
r.Cp.ShouldBeNull();
|
|
await host.DisposeAsync();
|
|
}
|
|
|
|
[Fact]
|
|
public async Task End_only_events_read_pages_backward_from_EndTime()
|
|
{
|
|
// StartTime unset + EndTime + NumValuesPerNode > 0 => backward from EndTime (Part 11; mirrors raw). The reverse
|
|
// window is (.., EndTime], so an event exactly AT EndTime is the first one returned.
|
|
var (host, nm, node, _) = await SetupAsync(Distinct(25));
|
|
var ids = DrainAsync(nm, node, max: 10, start: DateTime.MinValue, end: Epoch.AddSeconds(20));
|
|
ids.ShouldBe(Enumerable.Range(0, 21).Reverse().Select(i => $"e{i:0000}"));
|
|
await host.DisposeAsync();
|
|
}
|
|
|
|
[Fact]
|
|
public async Task A_backend_without_reverse_support_walks_a_reverse_read_back_never_answering_the_earliest_events()
|
|
{
|
|
// #691.4 (was refused BadHistoryOperationUnsupported since #675): served newest-first from forward reads.
|
|
var (host, nm, node, src) = await SetupAsync(Distinct(10));
|
|
var fwd = new ForwardOnlyEventSource(src);
|
|
nm.HistorianDataSource = fwd;
|
|
var r = ReadEvents(nm, node, max: 3, inboundCp: null, end: Early, start: Late);
|
|
r.Error.StatusCode.Code.ShouldBe(StatusCodes.Good);
|
|
r.Ids.ShouldBe(["e0009", "e0008", "e0007"]);
|
|
r.Cp.ShouldNotBeNull();
|
|
// Forward reads on the same backend are unaffected.
|
|
DrainAsync(nm, node, max: 4).ShouldBe(src.All.Select(e => e.EventId));
|
|
await host.DisposeAsync();
|
|
}
|
|
|
|
private static readonly DateTime Early = Epoch.AddDays(-1);
|
|
private static readonly DateTime Late = Epoch.AddDays(1);
|
|
|
|
// --- #675 review follow-ups: the reverse-events capability ------------------------------------------------
|
|
|
|
// --- #691.4: reverse events on a backend that reads forward only (the Gateway) are walked back ------------------
|
|
|
|
[Fact]
|
|
public async Task End_only_capped_read_on_a_forward_only_backend_pages_backward_from_EndTime()
|
|
{
|
|
// Part 11 (as raw, #686): the latest events at or before EndTime, newest first — on Gateway this used to keep its
|
|
// pre-#675 forward [MinValue, EndTime) meaning, i.e. the OLDEST events. The backend is never sent a reverse read.
|
|
var (host, nm, node, _) = await SetupAsync([]);
|
|
var gw = new GatewayShapedEventSource(Distinct(25));
|
|
nm.HistorianDataSource = gw;
|
|
var ids = DrainAsync(nm, node, max: 10, start: DateTime.MinValue, end: Epoch.AddSeconds(20));
|
|
ids.ShouldBe(Enumerable.Range(0, 21).Reverse().Select(i => $"e{i:0000}"));
|
|
gw.ReverseCalls.ShouldBe(0);
|
|
gw.BadRequests.ShouldBe(0);
|
|
await host.DisposeAsync();
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Explicit_reverse_read_on_a_forward_only_backend_pages_the_reverse_window_newest_first()
|
|
{
|
|
// (EndTime, StartTime]: StartTime inclusive, EndTime exclusive.
|
|
var (host, nm, node, _) = await SetupAsync([]);
|
|
var gw = new GatewayShapedEventSource(Distinct(30));
|
|
nm.HistorianDataSource = gw;
|
|
var ids = DrainAsync(nm, node, max: 4, start: Epoch.AddSeconds(25), end: Epoch.AddSeconds(5));
|
|
ids.ShouldBe(Enumerable.Range(6, 20).Reverse().Select(i => $"e{i:0000}"));
|
|
gw.ReverseCalls.ShouldBe(0);
|
|
await host.DisposeAsync();
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Reverse_ties_on_a_forward_only_backend_page_in_a_stable_order_although_the_backend_shuffles_them()
|
|
{
|
|
// The gateway orders events by EventTimeUtc only, so ties come back in any order — this fake shuffles them on every
|
|
// call. The walk sorts ties by EventId (descending, newest-first) on every read, and the oversized-cluster over-fetch
|
|
// does the same, so a boundary-tie skip always lines up. Cap 3 cuts through a 7-way cluster.
|
|
var events = new List<HistoricalEvent> { Ev("a0", 0) };
|
|
for (var i = 0; i < 7; i++) events.Add(Ev($"t{i}", 5));
|
|
events.Add(Ev("z0", 9));
|
|
var (host, nm, node, _) = await SetupAsync([]);
|
|
nm.HistorianDataSource = new GatewayShapedEventSource(events, shuffleSeed: 7);
|
|
DrainAsync(nm, node, max: 3, start: DateTime.MinValue, end: Epoch.AddSeconds(30)).ShouldBe(
|
|
["z0", "t6", "t5", "t4", "t3", "t2", "t1", "t0", "a0"]);
|
|
await host.DisposeAsync();
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Uncapped_reverse_read_on_a_forward_only_backend_is_one_forward_read()
|
|
{
|
|
var (host, nm, node, _) = await SetupAsync([]);
|
|
var gw = new GatewayShapedEventSource(Distinct(30));
|
|
nm.HistorianDataSource = gw;
|
|
var r = ReadEvents(nm, node, max: 0, inboundCp: null, end: Epoch.AddSeconds(5), start: Epoch.AddSeconds(25));
|
|
r.Ids.ShouldBe(Enumerable.Range(6, 20).Reverse().Select(i => $"e{i:0000}"));
|
|
r.Cp.ShouldBeNull();
|
|
gw.ForwardCalls.ShouldBe(1);
|
|
await host.DisposeAsync();
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Reverse_paging_on_a_forward_only_backend_is_exact_on_random_data()
|
|
{
|
|
var (host, nm, node, _) = await SetupAsync([]);
|
|
for (var seed = 0; seed < 80; seed++)
|
|
{
|
|
var rng = new Random(seed);
|
|
var events = new List<HistoricalEvent>();
|
|
var t = Epoch.Ticks;
|
|
var n = rng.Next(1, 120);
|
|
while (events.Count < n)
|
|
{
|
|
t += rng.Next(4) 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(5) == 0 ? rng.Next(2, 8) : 1;
|
|
for (var k = 0; k < ties; k++)
|
|
{
|
|
var at = new DateTime(t, DateTimeKind.Utc);
|
|
events.Add(new HistoricalEvent($"{Guid.NewGuid():N}", "src", at, at, "m", 500));
|
|
}
|
|
}
|
|
|
|
nm.HistorianDataSource = new GatewayShapedEventSource(events, shuffleSeed: seed);
|
|
nm.MaxTieClusterOverfetch = rng.Next(2) == 0 ? 8 : 65536;
|
|
var max = (uint)rng.Next(1, 9);
|
|
var end = new DateTime(t + rng.Next(0, 3) * TimeSpan.TicksPerSecond, DateTimeKind.Utc);
|
|
var expected = events.OrderByDescending(e => e.EventTimeUtc).ThenByDescending(e => e.EventId, StringComparer.Ordinal)
|
|
.Select(e => e.EventId).ToList();
|
|
|
|
var all = new List<string>();
|
|
byte[]? cp = null;
|
|
var pages = 0;
|
|
do
|
|
{
|
|
var r = ReadEvents(nm, node, max, cp, end, DateTime.MinValue);
|
|
r.Error.StatusCode.Code.ShouldBe(StatusCodes.Good, $"seed {seed}");
|
|
all.AddRange(r.Ids);
|
|
cp = r.Cp;
|
|
(++pages).ShouldBeLessThan(2000);
|
|
}
|
|
while (cp is not null);
|
|
|
|
all.ShouldBe(expected, $"seed {seed}, N = {max}");
|
|
}
|
|
|
|
await host.DisposeAsync();
|
|
}
|
|
|
|
[Fact]
|
|
public async Task A_NotSupportedException_from_inside_a_supporting_backend_is_a_backend_fault_not_unsupported()
|
|
{
|
|
// A NotSupportedException thrown by a driver library (Npgsql, SqlClient) or the projection is a fault: it must be
|
|
// logged as an Error like any other backend failure, never downgraded to the "backend cannot read reverse" Warning.
|
|
var logger = new CapturingLogger();
|
|
var (host, nm, node, _) = await SetupAsync(Distinct(10), logger: logger);
|
|
nm.HistorianDataSource = new ThrowingReverseSource();
|
|
ReadEvents(nm, node, max: 3, inboundCp: null, end: Early, start: Late).Error.StatusCode.Code
|
|
.ShouldBe(StatusCodes.BadHistoryOperationUnsupported);
|
|
logger.Entries.ShouldContain(e => e.Level == LogLevel.Error && e.Exception is NotSupportedException);
|
|
logger.Entries.ShouldNotContain(e => e.Level == LogLevel.Warning);
|
|
await host.DisposeAsync();
|
|
}
|
|
|
|
/// <summary>
|
|
/// Shaped like the Gateway's SQL event read (#691.4): forward only (<see cref="SupportsReverseEvents"/> false — a
|
|
/// reverse call is counted and refused), <c>[start, end]</c> INCLUSIVE, ordered by event time only with ties in an
|
|
/// arbitrary order (shuffled on every call when a seed is given), capped with a truncation flag.
|
|
/// </summary>
|
|
private sealed class GatewayShapedEventSource(List<HistoricalEvent> events, int? shuffleSeed = null) : IHistorianDataSource
|
|
{
|
|
private readonly Random? _rng = shuffleSeed is { } s ? new Random(s) : null;
|
|
public int ReverseCalls { get; private set; }
|
|
public int ForwardCalls { get; private set; }
|
|
public int BadRequests { get; private set; }
|
|
public bool SupportsReverseEvents => false;
|
|
public Task<HistoricalEventsResult> ReadEventsAsync(string? n, DateTime s, DateTime e, int m, CancellationToken ct)
|
|
{
|
|
ForwardCalls++;
|
|
if (s > e || s < DateTime.UnixEpoch) BadRequests++;
|
|
var window = events.Where(x => x.EventTimeUtc >= s && x.EventTimeUtc <= e)
|
|
.OrderBy(x => x.EventTimeUtc).ThenBy(_ => _rng?.Next() ?? 0).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)
|
|
{
|
|
if (reverse) { ReverseCalls++; throw new NotSupportedException("forward only"); }
|
|
return 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() { }
|
|
}
|
|
|
|
/// <summary>Supports reverse, but its read fails inside with a library-level NotSupportedException.</summary>
|
|
private sealed class ThrowingReverseSource : IHistorianDataSource
|
|
{
|
|
public bool SupportsReverseEvents => true;
|
|
public Task<HistoricalEventsResult> ReadEventsAsync(string? n, DateTime s, DateTime e, int m, CancellationToken ct) =>
|
|
throw new NotSupportedException("driver: unsupported wire type");
|
|
public Task<HistoricalEventsResult> ReadEventsAsync(string? n, DateTime s, DateTime e, int m, bool reverse, CancellationToken ct) =>
|
|
throw new NotSupportedException("driver: unsupported wire type");
|
|
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 CapturingLogger : ILogger<OtOpcUaNodeManager>
|
|
{
|
|
public System.Collections.Concurrent.ConcurrentQueue<(LogLevel Level, Exception? Exception, string Message)> Entries { get; } = new();
|
|
public IDisposable? BeginScope<TState>(TState state) where TState : notnull => null;
|
|
public bool IsEnabled(LogLevel logLevel) => true;
|
|
public void Log<TState>(LogLevel logLevel, EventId eventId, TState state, Exception? exception, Func<TState, Exception?, string> formatter) =>
|
|
Entries.Enqueue((logLevel, exception, formatter(state, exception)));
|
|
}
|
|
|
|
/// <summary>A backend that implements only the five-argument (forward) event read - Gateway / Null shaped.</summary>
|
|
private sealed class ForwardOnlyEventSource(IHistorianDataSource inner) : IHistorianDataSource
|
|
{
|
|
public Task<HistoricalEventsResult> ReadEventsAsync(string? n, DateTime s, DateTime e, int m, CancellationToken ct) =>
|
|
inner.ReadEventsAsync(n, s, e, m, ct);
|
|
public Task<HistorianRead> ReadRawAsync(string r, DateTime s, DateTime e, uint m, CancellationToken ct) =>
|
|
inner.ReadRawAsync(r, s, e, m, ct);
|
|
public Task<HistorianRead> ReadProcessedAsync(string r, DateTime s, DateTime e, TimeSpan i, HistoryAggregateType a, CancellationToken ct) =>
|
|
inner.ReadProcessedAsync(r, s, e, i, a, ct);
|
|
public Task<HistorianRead> ReadAtTimeAsync(string r, IReadOnlyList<DateTime> t, CancellationToken ct) =>
|
|
inner.ReadAtTimeAsync(r, t, ct);
|
|
public HistorianHealthSnapshot GetHealthSnapshot() => inner.GetHealthSnapshot();
|
|
public void Dispose() { }
|
|
}
|
|
|
|
private sealed class RawSeriesSource : IHistorianDataSource
|
|
{
|
|
public Task<HistorianRead> ReadRawAsync(string r, DateTime s, DateTime e, uint m, CancellationToken ct)
|
|
{
|
|
var rows = Enumerable.Range(0, 10).Select(i => new DataValueSnapshot((double)i, StatusCodes.Good,
|
|
Epoch.AddSeconds(i), Epoch.AddSeconds(i))).Where(v => v.SourceTimestampUtc >= s && v.SourceTimestampUtc <= e)
|
|
.Take((int)m).ToList();
|
|
return Task.FromResult(new HistorianRead(rows, null));
|
|
}
|
|
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 Task<HistoricalEventsResult> ReadEventsAsync(string? n, DateTime s, DateTime e, int m, CancellationToken ct) =>
|
|
NullHistorianDataSource.Instance.ReadEventsAsync(n, s, e, m, ct);
|
|
public HistorianHealthSnapshot GetHealthSnapshot() => NullHistorianDataSource.Instance.GetHealthSnapshot();
|
|
public void Dispose() { }
|
|
}
|
|
|
|
// --- helpers ---------------------------------------------------------------------------------
|
|
|
|
private static HistoricalEvent Ev(string id, int seconds) =>
|
|
new(id, "src", Epoch.AddSeconds(seconds), Epoch.AddSeconds(seconds), "m", 500);
|
|
|
|
private static List<HistoricalEvent> Distinct(int n) =>
|
|
Enumerable.Range(0, n).Select(i => Ev($"e{i:0000}", i)).ToList();
|
|
|
|
private static NodeId SecondNode(OtOpcUaNodeManager nm)
|
|
{
|
|
nm.EnsureFolder("eq-evt2", parentNodeId: null, displayName: "Equipment2", realm: AddressSpaceRealm.Uns);
|
|
nm.MaterialiseAlarmCondition("alarm-2", "eq-evt2", "HighTemp", "OffNormalAlarm", severity: 700, realm: AddressSpaceRealm.Uns);
|
|
return nm.TryGetFolder("eq-evt2")!.NodeId;
|
|
}
|
|
|
|
private async Task<(OpcUaApplicationHost Host, OtOpcUaNodeManager Nm, NodeId Node, SqlLikeEventSource Src)> SetupAsync(
|
|
List<HistoricalEvent> events, long resolution = 1, ILogger<OtOpcUaNodeManager>? logger = null)
|
|
{
|
|
var applicationUri = $"urn:OtOpcUa.EventPagingTest:{Guid.NewGuid():N}";
|
|
var started = await OpcUaHostStart.StartAsync(
|
|
port => new OpcUaApplicationHost(
|
|
new OpcUaApplicationHostOptions
|
|
{
|
|
ApplicationName = "OtOpcUa.EventPagingTest",
|
|
ApplicationUri = applicationUri,
|
|
OpcUaPort = port,
|
|
PublicHostname = "localhost",
|
|
PkiStoreRoot = _pkiRoot,
|
|
},
|
|
NullLogger<OpcUaApplicationHost>.Instance),
|
|
() => new OtOpcUaSdkServer(logger), Ct);
|
|
var nm = started.Server.NodeManager!;
|
|
nm.HistoryContinuationStore = new InMemoryHistoryContinuationStore();
|
|
var src = new SqlLikeEventSource(events, resolution);
|
|
nm.HistorianDataSource = src;
|
|
|
|
const string equipmentId = "eq-evt";
|
|
nm.EnsureFolder(equipmentId, parentNodeId: null, displayName: "Equipment", realm: AddressSpaceRealm.Uns);
|
|
nm.MaterialiseAlarmCondition("alarm-1", equipmentId, "HighTemp", "OffNormalAlarm", severity: 700, realm: AddressSpaceRealm.Uns);
|
|
return (started.Host, nm, nm.TryGetFolder(equipmentId)!.NodeId, src);
|
|
}
|
|
|
|
private static List<string> DrainAsync(
|
|
OtOpcUaNodeManager nm, NodeId node, uint max, DateTime? end = null, DateTime? start = null, List<int>? pageSizes = null)
|
|
{
|
|
var all = new List<string>();
|
|
byte[]? cp = null;
|
|
var pages = 0;
|
|
do
|
|
{
|
|
var r = ReadEvents(nm, node, max, cp, end, start);
|
|
r.Error.StatusCode.Code.ShouldBe(StatusCodes.Good);
|
|
all.AddRange(r.Ids);
|
|
pageSizes?.Add(r.Ids.Count);
|
|
cp = r.Cp;
|
|
(++pages).ShouldBeLessThan(60, "paging must terminate");
|
|
}
|
|
while (cp is not null);
|
|
return all;
|
|
}
|
|
|
|
private static (List<string> Ids, ServiceResult Error, byte[]? Cp) ReadEvents(
|
|
OtOpcUaNodeManager nm, NodeId node, uint max, byte[]? inboundCp, DateTime? end = null, DateTime? start = null) =>
|
|
ReadBatch(nm, max, end, start, (node, inboundCp))[0];
|
|
|
|
private static List<(List<string> Ids, ServiceResult Error, byte[]? Cp)> ReadBatch(
|
|
OtOpcUaNodeManager nm, uint max, DateTime? end, params (NodeId Node, byte[]? Cp)[] reads) =>
|
|
ReadBatch(nm, max, end, null, reads);
|
|
|
|
private static List<(List<string> Ids, ServiceResult Error, byte[]? Cp)> ReadBatch(
|
|
OtOpcUaNodeManager nm, uint max, DateTime? end, DateTime? start, params (NodeId Node, byte[]? Cp)[] reads)
|
|
{
|
|
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 ?? Epoch.AddDays(-1),
|
|
EndTime = end ?? Epoch.AddDays(1),
|
|
NumValuesPerNode = max,
|
|
Filter = filter,
|
|
};
|
|
var nodesToRead = reads.Select(r => new HistoryReadValueId { NodeId = r.Node, ContinuationPoint = r.Cp }).ToList();
|
|
var results = reads.Select(_ => (SdkHistoryReadResult)null!).ToList();
|
|
var errors = reads.Select(_ => (ServiceResult)null!).ToList();
|
|
nm.HistoryRead(context, details, TimestampsToReturn.Both, false, nodesToRead, results, errors);
|
|
|
|
var all = new List<(List<string>, ServiceResult, byte[]?)>();
|
|
for (var i = 0; i < reads.Length; i++)
|
|
{
|
|
var ids = new List<string>();
|
|
if (results[i]?.HistoryData is { } ho && !ExtensionObject.IsNull(ho))
|
|
{
|
|
var he = (HistoryEvent)ExtensionObject.ToEncodeable(ho);
|
|
foreach (var e in he.Events)
|
|
ids.Add(System.Text.Encoding.UTF8.GetString((byte[])e.EventFields[0].Value));
|
|
}
|
|
|
|
all.Add((ids, errors[i], results[i]?.ContinuationPoint));
|
|
}
|
|
|
|
return all;
|
|
}
|
|
|
|
private static void InvokeRelease(OtOpcUaNodeManager nm, NodeId node, byte[] cp)
|
|
{
|
|
var context = new OperationContext(
|
|
new RequestHeader(), secureChannelContext: null, RequestType.HistoryRead, identity: null);
|
|
var nodesToRead = new List<HistoryReadValueId> { new() { NodeId = node, ContinuationPoint = cp } };
|
|
var results = new List<SdkHistoryReadResult> { null! };
|
|
var errors = new List<ServiceResult> { null! };
|
|
nm.HistoryRead(context, new ReadEventDetails { StartTime = Epoch, EndTime = Epoch.AddDays(1), NumValuesPerNode = 4 },
|
|
TimestampsToReturn.Both, true, nodesToRead, results, errors);
|
|
}
|
|
|
|
/// <summary>Behaves like the SqlServer/QuestDB event reads: (time, id) order, [start, end), cap, truncation flag.</summary>
|
|
private sealed class SqlLikeEventSource(List<HistoricalEvent> events, long resolution = 1) : IHistorianDataSource
|
|
{
|
|
public long TimestampResolutionTicks => resolution;
|
|
public bool SupportsReverseEvents => true;
|
|
private DateTime Trunc(DateTime d) => d.AddTicks(-(d.Ticks % resolution));
|
|
|
|
public IReadOnlyList<HistoricalEvent> All { get; } =
|
|
events.OrderBy(e => e.EventTimeUtc).ThenBy(e => e.EventId, StringComparer.Ordinal).ToList();
|
|
public int Calls { get; private set; }
|
|
|
|
public Task<HistoricalEventsResult> ReadEventsAsync(
|
|
string? sourceName, DateTime startUtc, DateTime endUtc, int maxEvents, CancellationToken ct)
|
|
{
|
|
Calls++;
|
|
// Like the SQL backends: a reversed window is swapped, inputs truncate to the storage resolution.
|
|
if (startUtc > endUtc) (startUtc, endUtc) = (endUtc, startUtc);
|
|
startUtc = Trunc(startUtc);
|
|
endUtc = Trunc(endUtc);
|
|
var window = All.Where(e => e.EventTimeUtc >= startUtc && e.EventTimeUtc < endUtc).ToList();
|
|
var cap = maxEvents <= 0 ? 1000 : maxEvents;
|
|
var truncated = window.Count > cap;
|
|
return Task.FromResult(new HistoricalEventsResult(
|
|
window.Take(cap).ToList(), truncated ? new byte[] { 1 } : null));
|
|
}
|
|
|
|
/// <summary>Like the SQL backends' #675 overload: reverse => <c>(lo, hi]</c> newest-first by (time, id) descending.</summary>
|
|
public Task<HistoricalEventsResult> ReadEventsAsync(
|
|
string? sourceName, DateTime startUtc, DateTime endUtc, int maxEvents, bool reverse, CancellationToken ct)
|
|
{
|
|
if (!reverse) return ReadEventsAsync(sourceName, startUtc, endUtc, maxEvents, ct);
|
|
Calls++;
|
|
var (lo, hi) = startUtc <= endUtc ? (startUtc, endUtc) : (endUtc, startUtc);
|
|
lo = Trunc(lo);
|
|
hi = Trunc(hi);
|
|
var window = All.Where(e => e.EventTimeUtc > lo && e.EventTimeUtc <= hi).Reverse().ToList();
|
|
var cap = maxEvents <= 0 ? 1000 : maxEvents;
|
|
var truncated = window.Count > cap;
|
|
return Task.FromResult(new HistoricalEventsResult(
|
|
window.Take(cap).ToList(), truncated ? new byte[] { 1 } : null));
|
|
}
|
|
|
|
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() { }
|
|
}
|
|
|
|
public void Dispose()
|
|
{
|
|
if (Directory.Exists(_pkiRoot))
|
|
{
|
|
try { Directory.Delete(_pkiRoot, recursive: true); } catch { /* best effort */ }
|
|
}
|
|
}
|
|
}
|