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

Merge feat/675-reverse-event-paging: reverse event HistoryRead paging, end-only backward, Gateway MaxValue pin (#675)

Browse Source
This commit is contained in:
Joseph Doherty
2026-10-11 11:23:51 -04:00
parent e59ab69398 8d154c6a81
commit 8e6ac4b344
20 changed files with 752 additions and 60 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/EventHistoryPagingOverTheWireTests.cs
+55 -3
View File
@@ -54,8 +54,48 @@ public sealed class EventHistoryPagingOverTheWireTests
finally { await rig.DisposeAsync(); }
}
[Fact]
public async Task A_capped_REVERSE_event_read_pages_newest_first_through_the_session_store_exactly_once()
{
// Gitea #675: StartTime later than EndTime. Before the fix the cap kept the EARLIEST 10, ascending, with no
// continuation point.
var ct = TestContext.Current.CancellationToken;
var rig = await StartAsync("EvtPagingWireRev", encrypted: false, ct);
try
{
var events = Enumerable.Range(0, 25)
.Select(i => new HistoricalEvent($"e{i:00}", "src", Epoch.AddSeconds(i / 2), Epoch, "m", 500))
.ToList(); // pairs share a timestamp, so page boundaries cut through ties
rig.Nm.HistorianDataSource = new SqlLikeEventSource(events);
rig.MaterialiseNative();
rig.Enforce(false, [Key()]);
using var session = await rig.ConnectAsync(anonymous: true, ct);
var node = new NodeId(RawDeviceFolder, (ushort)session.NamespaceUris.GetIndex(V3NodeIds.RawNamespaceUri));
var collected = new List<string>();
var sizes = new List<int>();
byte[]? cp = null;
do
{
var (ids, next) = await ReadPageAsync(session, node, cp, ct, reverse: true);
collected.AddRange(ids);
sizes.Add(ids.Count);
cp = next;
sizes.Count.ShouldBeLessThan(30, "paging must terminate");
}
while (cp is not null);
sizes.ShouldBe([10, 10, 5]);
// Newest-first: time descending, ties by EventId descending.
collected.ShouldBe(events.OrderByDescending(e => e.EventTimeUtc)
.ThenByDescending(e => e.EventId, StringComparer.Ordinal).Select(e => e.EventId));
}
finally { await rig.DisposeAsync(); }
}
private static async Task<(List<string> Ids, byte[]? Cp)> ReadPageAsync(
ISession session, NodeId node, byte[]? cp, CancellationToken ct)
ISession session, NodeId node, byte[]? cp, CancellationToken ct, bool reverse = false)
{
var filter = new EventFilter();
filter.SelectClauses.Add(new SimpleAttributeOperand(ObjectTypeIds.BaseEventType, new QualifiedName("EventId"))
@@ -64,8 +104,8 @@ public sealed class EventHistoryPagingOverTheWireTests
});
var details = new ReadEventDetails
{
StartTime = Epoch.AddDays(-1),
EndTime = Epoch.AddDays(1),
StartTime = reverse ? Epoch.AddDays(1) : Epoch.AddDays(-1),
EndTime = reverse ? Epoch.AddDays(-1) : Epoch.AddDays(1),
NumValuesPerNode = 10,
Filter = filter,
};
@@ -94,6 +134,18 @@ public sealed class EventHistoryPagingOverTheWireTests
return new HistoricalEventsResult(window.Take(cap).ToList(), window.Count > cap ? new byte[] { 1 } : null);
}
/// <summary>The #675 overload: reverse is <c>(lo, hi]</c> by (time, id) descending.</summary>
public async Task<HistoricalEventsResult> ReadEventsAsync(
string? sourceName, DateTime startUtc, DateTime endUtc, int maxEvents, bool reverse, CancellationToken ct)
{
if (!reverse) return await ReadEventsAsync(sourceName, startUtc, endUtc, maxEvents, ct).ConfigureAwait(false);
await Task.Delay(1, ct).ConfigureAwait(false);
var (lo, hi) = startUtc <= endUtc ? (startUtc, endUtc) : (endUtc, startUtc);
var window = _all.Where(e => e.EventTimeUtc > lo && e.EventTimeUtc <= hi).Reverse().ToList();
var cap = maxEvents <= 0 ? 1000 : maxEvents;
return new HistoricalEventsResult(window.Take(cap).ToList(), window.Count > cap ? new byte[] { 1 } : null);
}
public Task<CoreHistoryReadResult> ReadRawAsync(string r, DateTime s, DateTime e, uint m, CancellationToken ct) =>
NullHistorianDataSource.Instance.ReadRawAsync(r, s, e, m, ct);
public Task<CoreHistoryReadResult> ReadProcessedAsync(
tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadEventsPagingTests.cs
+186 -7
View File
@@ -199,11 +199,168 @@ public sealed class NodeManagerHistoryReadEventsPagingTests : IDisposable
var mixed = ReadBatch(nm, 4, null, (node, good1[0].Cp), (other, Guid.NewGuid().ToByteArray()));
mixed[0].Error.StatusCode.Code.ShouldBe(StatusCodes.Good);
mixed[0].Ids.ShouldBe(src.All.Skip(4).Take(3).Select(e => e.EventId));
// 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.BadHistoryOperationUnsupported);
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_answers_BadHistoryOperationUnsupported_not_the_earliest_events()
{
var (host, nm, node, src) = await SetupAsync(Distinct(10));
nm.HistorianDataSource = new ForwardOnlyEventSource(src);
ReadEvents(nm, node, max: 3, inboundCp: null, end: Early, start: Late).Error.StatusCode.Code
.ShouldBe(StatusCodes.BadHistoryOperationUnsupported);
// 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);
/// <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)
@@ -265,16 +422,18 @@ public sealed class NodeManagerHistoryReadEventsPagingTests : IDisposable
return (started.Host, nm, nm.TryGetFolder(equipmentId)!.NodeId, src);
}
private static List<string> DrainAsync(OtOpcUaNodeManager nm, NodeId node, uint max, DateTime? end = null)
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);
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");
}
@@ -283,11 +442,15 @@ public sealed class NodeManagerHistoryReadEventsPagingTests : IDisposable
}
private static (List<string> Ids, ServiceResult Error, byte[]? Cp) ReadEvents(
OtOpcUaNodeManager nm, NodeId node, uint max, byte[]? inboundCp, DateTime? end = null) =>
ReadBatch(nm, max, end, (node, inboundCp))[0];
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)
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);
@@ -298,7 +461,7 @@ public sealed class NodeManagerHistoryReadEventsPagingTests : IDisposable
});
var details = new ReadEventDetails
{
StartTime = Epoch.AddDays(-1),
StartTime = start ?? Epoch.AddDays(-1),
EndTime = end ?? Epoch.AddDays(1),
NumValuesPerNode = max,
Filter = filter,
@@ -361,6 +524,22 @@ public sealed class NodeManagerHistoryReadEventsPagingTests : IDisposable
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(