From f9fd495292d9d9c7f919d2f2844b71ae28654260 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Sun, 11 Oct 2026 08:21:02 -0400 Subject: [PATCH 1/3] fix(opcua): issue and resume continuation points for event HistoryRead (#669) A capped events read now pages: the source's truncation flag triggers a session-bound cursor (time inclusive + tie skip), oversized tie clusters are paged within via a bounded half-open over-fetch, and release/unknown/ cross-kind points behave as for Raw. --- docs/Historian.md | 21 +- .../HistoryPaging.cs | 9 +- .../OtOpcUaNodeManager.cs | 123 +++++++-- .../EventHistoryPagingOverTheWireTests.cs | 107 ++++++++ ...NodeManagerHistoryReadEventsPagingTests.cs | 258 ++++++++++++++++++ 5 files changed, 498 insertions(+), 20 deletions(-) create mode 100644 tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/EventHistoryPagingOverTheWireTests.cs create mode 100644 tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadEventsPagingTests.cs diff --git a/docs/Historian.md b/docs/Historian.md index 38d9bd37..93751a14 100644 --- a/docs/Historian.md +++ b/docs/Historian.md @@ -583,6 +583,23 @@ are disposed when the session closes). Resuming an unknown / evicted / released `BadContinuationPointInvalid`. `releaseContinuationPoints` drops the stored cursors without reading data. +### Continuation-point paging (Events) (#669) + +A capped event read (`NumValuesPerNode > 0`) pages exactly like Raw, with one difference: the trigger is the **source's +truncation flag** (`HistoricalEventsResult.ContinuationPoint != null`), never the page count, so a cap-sized page the source +says is complete carries no continuation point. `NumValuesPerNode == 0` stays unbounded and never pages. + +- The cursor is `(source, next start, end, boundary skip, cap)` in the same session-bound store as Raw (flagged as an events + cursor — a Raw point presented to an events read, or the reverse, is `BadContinuationPointInvalid`). The next page reads from + the last event's `EventTimeUtc` **inclusive** and drops the events at that timestamp already emitted. +- A tie cluster larger than the cap is paged within by over-fetching the half-open window `[T, T + resolution)` up to + `MaxTieClusterOverfetch`; beyond that bound the node answers `BadHistoryOperationUnsupported` (raise the bound). +- **Tie exactness depends on the backend.** SqlServer and QuestDB order by `(EventTimeUtc, EventId)`, so the skip count is exact. + The Gateway backend has no stable intra-timestamp order (and only a truncation flag), so ties at a page boundary are + best-effort there. +- Reverse event reads (`start > end`) are not paged server-side; no continuation point is issued for them. +- No session on the request (in-process callers) degrades to single-shot, as for Raw. + ### Total aggregate The OPC UA `Total` aggregate is **supported** over the HistorianGateway backend. The gateway exposes a @@ -725,8 +742,8 @@ result never has both. `[start, end)` by `EventTimeUtc`; **a reversed window (`end < start`) is swapped and returned ascending**. A null `sourceName` means all sources, otherwise `SourceName = @source`. Ordered `(EventTimeUtc, EventId)` ascending. A bounded read fetches `maxEvents + 1` rows and -returns a non-null continuation point when the extra row exists; the node manager does not currently issue or resume event -continuation points, so a client-capped read returns only the first page. +returns a non-null continuation point when the extra row exists; the node manager turns that flag into an OPC UA continuation +point (see [Continuation-point paging (Events)](#continuation-point-paging-events-669)). ### Aggregates (Processed reads) diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/HistoryPaging.cs b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/HistoryPaging.cs index 6fa8c12b..c4c4ad53 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/HistoryPaging.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/HistoryPaging.cs @@ -31,12 +31,19 @@ namespace ZB.MOM.WW.OtOpcUa.OpcUaServer; /// prior pages and must be dropped from the head of the next page (tie de-dup). /// /// The client's per-page cap; re-applied to every resumed page. +/// +/// true for an EVENT HistoryRead cursor (Gitea #669): then carries the +/// registered historian event-source name and the boundary fields are over EventTimeUtc. A cursor +/// is only ever resumed by the read kind that issued it, so a raw point presented to an events read +/// (or vice versa) is a BadContinuationPointInvalid miss. +/// internal sealed record HistoryContinuationState( string Tagname, DateTime NextStartUtc, DateTime EndUtc, int BoundarySkipCount, - uint NumValuesPerNode); + uint NumValuesPerNode, + bool IsEvents = false); /// /// Pure server-side continuation-point paging decisions for the count-capped variable-history arms diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs index 859b69b2..b570efb9 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs @@ -5086,6 +5086,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 { // Snapshot the select clauses once — the same filter projects every node's events. var selectClauses = details.Filter?.SelectClauses ?? new SimpleAttributeOperandCollection(); + var session = context.OperationContext?.Session; nodesToProcess = AuthorizeHistoryRead(context, nodesToProcess, results, errors); WithHistoryReadBatchLimiter(nodesToProcess, results, errors, () => @@ -5108,7 +5109,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 } var sn = sourceName; - work.Add(() => ServeEventsAsync(handle, sn, details, selectClauses, results, errors, ct)); + work.Add(() => ServeEventsAsync(handle, sn, details, selectClauses, nodesToRead, session, results, errors, ct)); } RunBounded(work, HistoryReadBatchParallelism); @@ -5134,32 +5135,120 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 string? sourceName, ReadEventDetails details, SimpleAttributeOperandCollection selectClauses, + IList nodesToRead, + ISession? session, IList results, IList errors, CancellationToken ct) { + var inboundCp = nodesToRead[handle.Index].ContinuationPoint; + var resumed = inboundCp is { Length: > 0 }; try { - // NOT under the node-manager Lock — awaiting the async source is safe here. - var sourceResult = await _historianDataSource.ReadEventsAsync( - sourceName, - details.StartTime, - details.EndTime, - // NumValuesPerNode==0 means "no limit — return ALL values" per OPC UA - // Part 4/11, so translate it to UNBOUNDED (int.MaxValue) here. Passing the int<=0 - // backend-default-cap sentinel instead would silently truncate a "give me everything" - // events read at the backend default. A positive cap passes through (clamped). - EventMaxEvents(details.NumValuesPerNode), - ct).ConfigureAwait(false); + var startUtc = details.StartTime; + var endUtc = details.EndTime; + var numValuesPerNode = details.NumValuesPerNode; + var boundarySkip = 0; - var historyEvent = ProjectEvents(sourceResult.Events, selectClauses); + if (resumed) + { + // Single-use take. A miss (unknown / evicted / malformed / a RAW cursor) ⇒ BadContinuationPointInvalid + // for THIS node only. + var state = _historyContinuationStore.TryTake(session, inboundCp!); + if (state is null || !state.IsEvents) + { + errors[handle.Index] = StatusCodes.BadContinuationPointInvalid; + results[handle.Index] = new SdkHistoryReadResult { StatusCode = StatusCodes.BadContinuationPointInvalid }; + return; + } + + sourceName = state.Tagname; + startUtc = state.NextStartUtc; + endUtc = state.EndUtc; + boundarySkip = state.BoundarySkipCount; + numValuesPerNode = state.NumValuesPerNode; + } + + // Reverse event reads (start > end) are not paged: the cursor below is forward-only. + var forward = !(details.EndTime != DateTime.MinValue && details.StartTime > details.EndTime); + + // NOT under the node-manager Lock — awaiting the async source is safe here. + // NumValuesPerNode==0 means "no limit — return ALL values" per OPC UA Part 4/11, so it is translated + // to UNBOUNDED (EventMaxEvents) rather than the backend's <=0 default-cap sentinel. + var sourceResult = await _historianDataSource.ReadEventsAsync( + sourceName, startUtc, endUtc, EventMaxEvents(numValuesPerNode), ct).ConfigureAwait(false); + + // Truncation is the SOURCE's word (non-null continuation point) — never inferred from the count, so a + // page that happens to be exactly cap-sized but complete never emits a point. + var truncated = sourceResult.ContinuationPoint is not null && numValuesPerNode > 0 && forward; + + IReadOnlyList events = sourceResult.Events; + if (resumed && boundarySkip > 0) + { + // Drop the events at the boundary timestamp already emitted on the prior page. + var i = 0; + while (i < events.Count && i < boundarySkip && events[i].EventTimeUtc == startUtc) i++; + if (i > 0) events = events.Skip(i).ToList(); + } + + byte[]? outboundCp = null; + + if (truncated && resumed + && (events.Count == 0 || events[^1].EventTimeUtc == startUtc)) + { + // Oversized tie cluster: more events share startUtc than the page cap, so (time, skip) alone + // cannot advance. Over-fetch the whole cluster over the half-open [T, T+resolution) window + // with an explicit bounded cap and page WITHIN it (same scheme as raw). + var resolution = Math.Max(1L, _historianDataSource.TimestampResolutionTicks); + var overfetchCap = (int)Math.Min(int.MaxValue - 1L, (long)Math.Max(1, MaxTieClusterOverfetch) + 1); + var cluster = (await _historianDataSource.ReadEventsAsync( + sourceName, startUtc, startUtc.AddTicks(resolution), overfetchCap, ct) + .ConfigureAwait(false)).Events; + + if (cluster.Count > MaxTieClusterOverfetch) + { + _logger.LogError( + "OtOpcUaNodeManager: HistoryReadEvents tie cluster at {StartUtc:O} for source '{Source}' has {Count} events, " + + "exceeding MaxTieClusterOverfetch={Max}; cannot page within it. Increase MaxTieClusterOverfetch.", + startUtc, sourceName, cluster.Count, MaxTieClusterOverfetch); + errors[handle.Index] = StatusCodes.BadHistoryOperationUnsupported; + results[handle.Index] = new SdkHistoryReadResult { StatusCode = StatusCodes.BadHistoryOperationUnsupported }; + return; + } + + HistoryPaging.SliceTieCluster( + cluster.Count, boundarySkip, numValuesPerNode, startUtc, endUtc, + out var sliceStart, out var sliceCount, out var nextStartUtc, out var nextSkip, + reverse: false, resolutionTicks: resolution); + + var slice = new List(sliceCount); + for (var i = sliceStart; i < sliceStart + sliceCount; i++) slice.Add(cluster[i]); + events = slice; + + if (nextStartUtc is { } resumeAt) + { + outboundCp = _historyContinuationStore.Save(session, new HistoryContinuationState( + sourceName ?? string.Empty, resumeAt, endUtc, nextSkip, numValuesPerNode, IsEvents: true)); + } + } + else if (truncated && events.Count > 0) + { + // Cursor from the last event: resume at its time INCLUSIVE, skipping the trailing ties emitted. + var boundary = events[^1].EventTimeUtc; + var skip = 0; + for (var i = events.Count - 1; i >= 0 && events[i].EventTimeUtc == boundary; i--) skip++; + outboundCp = _historyContinuationStore.Save(session, new HistoryContinuationState( + sourceName ?? string.Empty, boundary, endUtc, skip, numValuesPerNode, IsEvents: true)); + } + + var historyEvent = ProjectEvents(events, selectClauses); results[handle.Index] = new SdkHistoryReadResult { // No events ⇒ GoodNoData (the notifier is historized, the window just held no events). - StatusCode = sourceResult.Events.Count == 0 ? StatusCodes.GoodNoData : StatusCodes.Good, + StatusCode = events.Count == 0 ? StatusCodes.GoodNoData : StatusCodes.Good, HistoryData = new ExtensionObject(historyEvent), - // We never issue continuation points — every read returns the full window in one shot. - ContinuationPoint = null, + // Null when the page was complete, or when no session is available to hold the cursor. + ContinuationPoint = outboundCp, }; errors[handle.Index] = ServiceResult.Good; } @@ -5843,7 +5932,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 // Resume read: take the stored cursor. A miss (unknown / evicted / malformed point) ⇒ // BadContinuationPointInvalid for THIS node. var state = _historyContinuationStore.TryTake(session, inboundCp); - if (state is null) + if (state is null || state.IsEvents) { errors[handle.Index] = StatusCodes.BadContinuationPointInvalid; results[handle.Index] = new SdkHistoryReadResult { StatusCode = StatusCodes.BadContinuationPointInvalid }; diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/EventHistoryPagingOverTheWireTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/EventHistoryPagingOverTheWireTests.cs new file mode 100644 index 00000000..2ba2bd8f --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/EventHistoryPagingOverTheWireTests.cs @@ -0,0 +1,107 @@ +using Opc.Ua; +using Opc.Ua.Client; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Commons.OpcUa; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; +using CoreHistoryReadResult = ZB.MOM.WW.OtOpcUa.Core.Abstractions.HistoryReadResult; +using static ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests.ServerEventRig; + +namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests; + +/// +/// Gitea #669 over a real SDK client session: an event HistoryRead capped below the available events pages +/// through the PRODUCTION session-bound continuation store (SessionHistoryContinuationStore) — the +/// path the session-less unit tests cannot reach — and the union of pages is every event exactly once. +/// +public sealed class EventHistoryPagingOverTheWireTests +{ + private static readonly DateTime Epoch = new(2026, 10, 1, 0, 0, 0, DateTimeKind.Utc); + + [Fact] + public async Task A_capped_event_read_pages_through_the_session_store_exactly_once() + { + var ct = TestContext.Current.CancellationToken; + var rig = await StartAsync("EvtPagingWire", encrypted: false, ct); + try + { + // Source installed BEFORE the alarm materialises so the folder is promoted with the HistoryRead bit. + 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(); + byte[]? cp = null; + var pages = 0; + do + { + var (ids, next) = await ReadPageAsync(session, node, cp, ct); + collected.AddRange(ids); + cp = next; + (++pages).ShouldBeLessThan(30, "paging must terminate"); + } + while (cp is not null); + + pages.ShouldBeGreaterThan(2, "25 events at a cap of 10 must page over the wire"); + collected.ShouldBe(events.Select(e => e.EventId)); + } + finally { await rig.DisposeAsync(); } + } + + private static async Task<(List Ids, byte[]? Cp)> ReadPageAsync( + ISession session, NodeId node, byte[]? cp, CancellationToken ct) + { + var filter = new EventFilter(); + filter.SelectClauses.Add(new SimpleAttributeOperand(ObjectTypeIds.BaseEventType, new QualifiedName("EventId")) + { + AttributeId = Attributes.Value, + }); + var details = new ReadEventDetails + { + StartTime = Epoch.AddDays(-1), + EndTime = Epoch.AddDays(1), + NumValuesPerNode = 10, + Filter = filter, + }; + var response = await session.HistoryReadAsync( + null, new ExtensionObject(details), TimestampsToReturn.Both, false, + new HistoryReadValueIdCollection { new HistoryReadValueId { NodeId = node, ContinuationPoint = cp } }, ct); + var result = response.Results.ShouldHaveSingleItem(); + StatusCode.IsBad(result.StatusCode).ShouldBeFalse($"HistoryRead failed: {result.StatusCode}"); + var he = (HistoryEvent)ExtensionObject.ToEncodeable(result.HistoryData); + var ids = he.Events.Select(e => System.Text.Encoding.UTF8.GetString((byte[])e.EventFields[0].Value)).ToList(); + return (ids, result.ContinuationPoint is { Length: > 0 } c ? c : null); + } + + /// (time, id) order, [start, end), cap + truncation flag — the SQL backends' contract. + private sealed class SqlLikeEventSource(List events) : IHistorianDataSource + { + private readonly List _all = + events.OrderBy(e => e.EventTimeUtc).ThenBy(e => e.EventId, StringComparer.Ordinal).ToList(); + + public async Task ReadEventsAsync( + string? sourceName, DateTime startUtc, DateTime endUtc, int maxEvents, CancellationToken ct) + { + await Task.Delay(1, ct).ConfigureAwait(false); // complete on a pool thread, like a real backend + var window = _all.Where(e => e.EventTimeUtc >= startUtc && e.EventTimeUtc < endUtc).ToList(); + var cap = maxEvents <= 0 ? 1000 : maxEvents; + return new HistoricalEventsResult(window.Take(cap).ToList(), window.Count > cap ? new byte[] { 1 } : null); + } + + public Task ReadRawAsync(string r, DateTime s, DateTime e, uint m, CancellationToken ct) => + NullHistorianDataSource.Instance.ReadRawAsync(r, s, e, m, ct); + public Task ReadProcessedAsync( + string r, DateTime s, DateTime e, TimeSpan i, HistoryAggregateType a, CancellationToken ct) => + NullHistorianDataSource.Instance.ReadProcessedAsync(r, s, e, i, a, ct); + public Task ReadAtTimeAsync(string r, IReadOnlyList t, CancellationToken ct) => + NullHistorianDataSource.Instance.ReadAtTimeAsync(r, t, ct); + public HistorianHealthSnapshot GetHealthSnapshot() => NullHistorianDataSource.Instance.GetHealthSnapshot(); + public void Dispose() { } + } +} diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadEventsPagingTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadEventsPagingTests.cs new file mode 100644 index 00000000..2e62664a --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadEventsPagingTests.cs @@ -0,0 +1,258 @@ +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; + +/// +/// Gitea #669 — server-side continuation-point paging for HistoryRead-Events. Same harness as +/// : a real SDK server, a session-less in-process +/// HistoryRead and an , over a fake source that behaves like +/// the SQL backends (ordered by (time, id), half-open window, cap + truncation flag). +/// +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 + { + 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 { 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(); + 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.BadHistoryOperationUnsupported); + 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(); + } + + // --- helpers --------------------------------------------------------------------------------- + + private static HistoricalEvent Ev(string id, int seconds) => + new(id, "src", Epoch.AddSeconds(seconds), Epoch.AddSeconds(seconds), "m", 500); + + private static List Distinct(int n) => + Enumerable.Range(0, n).Select(i => Ev($"e{i:0000}", i)).ToList(); + + private async Task<(OpcUaApplicationHost Host, OtOpcUaNodeManager Nm, NodeId Node, SqlLikeEventSource Src)> SetupAsync( + List events) + { + 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.Instance), + () => new OtOpcUaSdkServer(), Ct); + var nm = started.Server.NodeManager!; + nm.HistoryContinuationStore = new InMemoryHistoryContinuationStore(); + var src = new SqlLikeEventSource(events); + 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 DrainAsync(OtOpcUaNodeManager nm, NodeId node, uint max) + { + var all = new List(); + byte[]? cp = null; + var pages = 0; + do + { + var r = ReadEvents(nm, node, max, cp); + r.Error.StatusCode.Code.ShouldBe(StatusCodes.Good); + all.AddRange(r.Ids); + cp = r.Cp; + (++pages).ShouldBeLessThan(60, "paging must terminate"); + } + while (cp is not null); + return all; + } + + private static (List Ids, ServiceResult Error, byte[]? Cp) ReadEvents( + OtOpcUaNodeManager nm, NodeId node, uint max, byte[]? inboundCp) + { + 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 = Epoch.AddDays(-1), + EndTime = Epoch.AddDays(1), + NumValuesPerNode = max, + Filter = filter, + }; + var nodesToRead = new List { new() { NodeId = node, ContinuationPoint = inboundCp } }; + var results = new List { null! }; + var errors = new List { null! }; + nm.HistoryRead(context, details, TimestampsToReturn.Both, false, nodesToRead, results, errors); + + var ids = new List(); + if (results[0]?.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)); + } + + return (ids, errors[0], results[0]?.ContinuationPoint); + } + + 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 { new() { NodeId = node, ContinuationPoint = cp } }; + var results = new List { null! }; + var errors = new List { null! }; + nm.HistoryRead(context, new ReadEventDetails { StartTime = Epoch, EndTime = Epoch.AddDays(1), NumValuesPerNode = 4 }, + TimestampsToReturn.Both, true, nodesToRead, results, errors); + } + + /// Behaves like the SqlServer/QuestDB event reads: (time, id) order, [start, end), cap, truncation flag. + private sealed class SqlLikeEventSource(List events) : IHistorianDataSource + { + public IReadOnlyList All { get; } = + events.OrderBy(e => e.EventTimeUtc).ThenBy(e => e.EventId, StringComparer.Ordinal).ToList(); + public int Calls { get; private set; } + + public Task ReadEventsAsync( + string? sourceName, DateTime startUtc, DateTime endUtc, int maxEvents, CancellationToken ct) + { + Calls++; + 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)); + } + + public Task ReadRawAsync(string r, DateTime s, DateTime e, uint m, CancellationToken ct) => + NullHistorianDataSource.Instance.ReadRawAsync(r, s, e, m, ct); + public Task ReadProcessedAsync( + string r, DateTime s, DateTime e, TimeSpan i, HistoryAggregateType a, CancellationToken ct) => + NullHistorianDataSource.Instance.ReadProcessedAsync(r, s, e, i, a, ct); + public Task ReadAtTimeAsync(string r, IReadOnlyList 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 */ } + } + } +} From acce99de9a49d69890e896a80e75e4ec620882d0 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Sun, 11 Oct 2026 08:30:36 -0400 Subject: [PATCH 2/3] fix(opcua): event paging review fixes (#669): open-ended end, node-bound cursors, save after projection --- .../OtOpcUaNodeManager.cs | 26 ++- ...NodeManagerHistoryReadEventsPagingTests.cs | 156 ++++++++++++++++-- 2 files changed, 158 insertions(+), 24 deletions(-) diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs index b570efb9..a40e1957 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs @@ -5155,7 +5155,8 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 // Single-use take. A miss (unknown / evicted / malformed / a RAW cursor) ⇒ BadContinuationPointInvalid // for THIS node only. var state = _historyContinuationStore.TryTake(session, inboundCp!); - if (state is null || !state.IsEvents) + // A point is bound to the node that issued it: another node's source name is a miss too. + if (state is null || !state.IsEvents || state.Tagname != (sourceName ?? string.Empty)) { errors[handle.Index] = StatusCodes.BadContinuationPointInvalid; results[handle.Index] = new SdkHistoryReadResult { StatusCode = StatusCodes.BadContinuationPointInvalid }; @@ -5169,8 +5170,13 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 numValuesPerNode = state.NumValuesPerNode; } + // Part 11: an unset EndTime (MinValue) with a StartTime means "page forward from StartTime". Normalise it + // to an open upper bound, or the cursor would store MinValue and a backend would swap the reversed + // resume window. A resumed read is always forward (the cursor is only ever stored for a forward read), + // so its direction comes from the cursor, not from the resume request's details. + if (!resumed && endUtc == DateTime.MinValue && startUtc != DateTime.MinValue) endUtc = DateTime.MaxValue; // Reverse event reads (start > end) are not paged: the cursor below is forward-only. - var forward = !(details.EndTime != DateTime.MinValue && details.StartTime > details.EndTime); + var forward = resumed || startUtc <= endUtc; // NOT under the node-manager Lock — awaiting the async source is safe here. // NumValuesPerNode==0 means "no limit — return ALL values" per OPC UA Part 4/11, so it is translated @@ -5191,7 +5197,9 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 if (i > 0) events = events.Skip(i).ToList(); } - byte[]? outboundCp = null; + // The cursor to store, if any. Saved only AFTER the page is projected, so a projection throw can + // never orphan a stored cursor. + HistoryContinuationState? pendingCursor = null; if (truncated && resumed && (events.Count == 0 || events[^1].EventTimeUtc == startUtc)) @@ -5227,8 +5235,8 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 if (nextStartUtc is { } resumeAt) { - outboundCp = _historyContinuationStore.Save(session, new HistoryContinuationState( - sourceName ?? string.Empty, resumeAt, endUtc, nextSkip, numValuesPerNode, IsEvents: true)); + pendingCursor = new HistoryContinuationState( + sourceName ?? string.Empty, resumeAt, endUtc, nextSkip, numValuesPerNode, IsEvents: true); } } else if (truncated && events.Count > 0) @@ -5237,11 +5245,12 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 var boundary = events[^1].EventTimeUtc; var skip = 0; for (var i = events.Count - 1; i >= 0 && events[i].EventTimeUtc == boundary; i--) skip++; - outboundCp = _historyContinuationStore.Save(session, new HistoryContinuationState( - sourceName ?? string.Empty, boundary, endUtc, skip, numValuesPerNode, IsEvents: true)); + pendingCursor = new HistoryContinuationState( + sourceName ?? string.Empty, boundary, endUtc, skip, numValuesPerNode, IsEvents: true); } var historyEvent = ProjectEvents(events, selectClauses); + var outboundCp = pendingCursor is null ? null : _historyContinuationStore.Save(session, pendingCursor); results[handle.Index] = new SdkHistoryReadResult { // No events ⇒ GoodNoData (the notifier is historized, the window just held no events). @@ -5932,7 +5941,8 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 // Resume read: take the stored cursor. A miss (unknown / evicted / malformed point) ⇒ // BadContinuationPointInvalid for THIS node. var state = _historyContinuationStore.TryTake(session, inboundCp); - if (state is null || state.IsEvents) + if (state is null || state.IsEvents + || !TryGetHistorizedTagname(handle.NodeId, out var boundTag) || boundTag != state.Tagname) { errors[handle.Index] = StatusCodes.BadContinuationPointInvalid; results[handle.Index] = new SdkHistoryReadResult { StatusCode = StatusCodes.BadContinuationPointInvalid }; diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadEventsPagingTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadEventsPagingTests.cs index 2e62664a..6cd34534 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadEventsPagingTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadEventsPagingTests.cs @@ -123,6 +123,106 @@ public sealed class NodeManagerHistoryReadEventsPagingTests : IDisposable 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 { null! }; + var errs = new List { null! }; + nm.HistoryRead(ctx, new ReadRawModifiedDetails { StartTime = Epoch, EndTime = Epoch.AddHours(1), NumValuesPerNode = 2, IsReadModified = false }, + TimestampsToReturn.Both, false, + new List { 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(); + 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); + mixed[0].Ids.ShouldBe(src.All.Skip(4).Take(3).Select(e => e.EventId)); + mixed[1].Error.StatusCode.Code.ShouldBe(StatusCodes.BadContinuationPointInvalid); + await host.DisposeAsync(); + } + + private sealed class RawSeriesSource : IHistorianDataSource + { + public Task 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 ReadProcessedAsync(string r, DateTime s, DateTime e, TimeSpan i, HistoryAggregateType a, CancellationToken ct) => + NullHistorianDataSource.Instance.ReadProcessedAsync(r, s, e, i, a, ct); + public Task ReadAtTimeAsync(string r, IReadOnlyList t, CancellationToken ct) => + NullHistorianDataSource.Instance.ReadAtTimeAsync(r, t, ct); + public Task 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) => @@ -131,8 +231,15 @@ public sealed class NodeManagerHistoryReadEventsPagingTests : IDisposable private static List 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 events) + List events, long resolution = 1) { var applicationUri = $"urn:OtOpcUa.EventPagingTest:{Guid.NewGuid():N}"; var started = await OpcUaHostStart.StartAsync( @@ -149,7 +256,7 @@ public sealed class NodeManagerHistoryReadEventsPagingTests : IDisposable () => new OtOpcUaSdkServer(), Ct); var nm = started.Server.NodeManager!; nm.HistoryContinuationStore = new InMemoryHistoryContinuationStore(); - var src = new SqlLikeEventSource(events); + var src = new SqlLikeEventSource(events, resolution); nm.HistorianDataSource = src; const string equipmentId = "eq-evt"; @@ -158,14 +265,14 @@ public sealed class NodeManagerHistoryReadEventsPagingTests : IDisposable return (started.Host, nm, nm.TryGetFolder(equipmentId)!.NodeId, src); } - private static List DrainAsync(OtOpcUaNodeManager nm, NodeId node, uint max) + private static List DrainAsync(OtOpcUaNodeManager nm, NodeId node, uint max, DateTime? end = null) { var all = new List(); byte[]? cp = null; var pages = 0; do { - var r = ReadEvents(nm, node, max, cp); + var r = ReadEvents(nm, node, max, cp, end); r.Error.StatusCode.Code.ShouldBe(StatusCodes.Good); all.AddRange(r.Ids); cp = r.Cp; @@ -176,7 +283,11 @@ public sealed class NodeManagerHistoryReadEventsPagingTests : IDisposable } private static (List Ids, ServiceResult Error, byte[]? Cp) ReadEvents( - OtOpcUaNodeManager nm, NodeId node, uint max, byte[]? inboundCp) + OtOpcUaNodeManager nm, NodeId node, uint max, byte[]? inboundCp, DateTime? end = null) => + ReadBatch(nm, max, end, (node, inboundCp))[0]; + + private static List<(List Ids, ServiceResult Error, byte[]? Cp)> ReadBatch( + OtOpcUaNodeManager nm, uint max, DateTime? end, params (NodeId Node, byte[]? Cp)[] reads) { var context = new OperationContext( new RequestHeader(), secureChannelContext: null, RequestType.HistoryRead, identity: null); @@ -188,24 +299,30 @@ public sealed class NodeManagerHistoryReadEventsPagingTests : IDisposable var details = new ReadEventDetails { StartTime = Epoch.AddDays(-1), - EndTime = Epoch.AddDays(1), + EndTime = end ?? Epoch.AddDays(1), NumValuesPerNode = max, Filter = filter, }; - var nodesToRead = new List { new() { NodeId = node, ContinuationPoint = inboundCp } }; - var results = new List { null! }; - var errors = new List { null! }; + 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 ids = new List(); - if (results[0]?.HistoryData is { } ho && !ExtensionObject.IsNull(ho)) + var all = new List<(List, ServiceResult, byte[]?)>(); + for (var i = 0; i < reads.Length; i++) { - var he = (HistoryEvent)ExtensionObject.ToEncodeable(ho); - foreach (var e in he.Events) - ids.Add(System.Text.Encoding.UTF8.GetString((byte[])e.EventFields[0].Value)); + var ids = new List(); + 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 (ids, errors[0], results[0]?.ContinuationPoint); + return all; } private static void InvokeRelease(OtOpcUaNodeManager nm, NodeId node, byte[] cp) @@ -220,8 +337,11 @@ public sealed class NodeManagerHistoryReadEventsPagingTests : IDisposable } /// Behaves like the SqlServer/QuestDB event reads: (time, id) order, [start, end), cap, truncation flag. - private sealed class SqlLikeEventSource(List events) : IHistorianDataSource + private sealed class SqlLikeEventSource(List events, long resolution = 1) : IHistorianDataSource { + public long TimestampResolutionTicks => resolution; + private DateTime Trunc(DateTime d) => d.AddTicks(-(d.Ticks % resolution)); + public IReadOnlyList All { get; } = events.OrderBy(e => e.EventTimeUtc).ThenBy(e => e.EventId, StringComparer.Ordinal).ToList(); public int Calls { get; private set; } @@ -230,6 +350,10 @@ public sealed class NodeManagerHistoryReadEventsPagingTests : IDisposable 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; From 50b504d29372c39e5683fa08d756afbfc593ef99 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Sun, 11 Oct 2026 08:37:24 -0400 Subject: [PATCH 3/3] fix(opcua): raw open-ended/end-only reads per Part 11; extreme-bound conformance (#669) --- .../OtOpcUaNodeManager.cs | 14 ++++- .../RawReadConformance.cs | 16 +++++ .../NodeManagerHistoryReadPagingTests.cs | 59 +++++++++++++++++++ 3 files changed, 88 insertions(+), 1 deletion(-) diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs index a40e1957..e9a9a123 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs @@ -5174,7 +5174,8 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 // to an open upper bound, or the cursor would store MinValue and a backend would swap the reversed // resume window. A resumed read is always forward (the cursor is only ever stored for a forward read), // so its direction comes from the cursor, not from the resume request's details. - if (!resumed && endUtc == DateTime.MinValue && startUtc != DateTime.MinValue) endUtc = DateTime.MaxValue; + if (!resumed && endUtc == DateTime.MinValue && startUtc != DateTime.MinValue) + endUtc = new DateTime(DateTime.MaxValue.Ticks, DateTimeKind.Utc); // Utc kind: a proto Timestamp conversion rejects Unspecified // Reverse event reads (start > end) are not paged: the cursor below is forward-only. var forward = resumed || startUtc <= endUtc; @@ -5966,6 +5967,17 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 tagname = resolved; startUtc = startTimeUtc; + + // Part 11: with NumValuesPerNode > 0 one bound may be left unset. Only a StartTime ⇒ page FORWARD from + // it to an open upper bound; only an EndTime ⇒ page BACKWARD from it (a reverse read, start > end). + // Without this an unset EndTime (MinValue) looked like a reverse read to a lower bound of "now". + if (numValuesPerNode > 0) + { + if (endUtc == DateTime.MinValue && startUtc != DateTime.MinValue) + endUtc = new DateTime(DateTime.MaxValue.Ticks, DateTimeKind.Utc); + else if (startUtc == DateTime.MinValue && endUtc != DateTime.MinValue) + (startUtc, endUtc) = (endUtc, new DateTime(0, DateTimeKind.Utc)); + } } // HistoryRead is NOT under the node-manager Lock — awaiting the async source is safe. diff --git a/tests/Drivers/Historian.Conformance/RawReadConformance.cs b/tests/Drivers/Historian.Conformance/RawReadConformance.cs index 07380824..41282291 100644 --- a/tests/Drivers/Historian.Conformance/RawReadConformance.cs +++ b/tests/Drivers/Historian.Conformance/RawReadConformance.cs @@ -40,6 +40,22 @@ public abstract class RawReadConformance : ConformanceBase Values(await B.CreateDataSource().ReadRawAsync("A", At(-1), At(100), uint.MaxValue, Ct)).Length.ShouldBe(5); } + /// + /// #669: the node manager now pages an open-ended read (Part 11: only a StartTime / only an EndTime) with + /// DateTime.MaxValue / DateTime.MinValue as the missing bound, so every backend must bind both + /// extremes without overflow, ±infinity or a cast error. + /// + [Fact] + public async Task Extreme_bounds_MaxValue_and_MinValue_are_safe() + { + await ConformanceData.SeedAAsync(B); + var max = new DateTime(DateTime.MaxValue.Ticks, DateTimeKind.Utc); + var min = new DateTime(0, DateTimeKind.Utc); + Values(await B.CreateDataSource().ReadRawAsync("A", At(10), max, 2, Ct)).ShouldBe([20d, 30d]); + Values(await B.CreateDataSource().ReadRawAsync("A", At(100), min, 2, Ct)).ShouldBe([50d, 40d]); + Values(await B.CreateDataSource().ReadRawAsync("A", At(10), max, 0, Ct)).ShouldBe([20d, 30d, 40d, 50d]); + } + [Fact] public async Task Reverse_read_is_descending_over_end_exclusive_start_inclusive() { diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadPagingTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadPagingTests.cs index 80785796..43ba469f 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadPagingTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadPagingTests.cs @@ -73,6 +73,65 @@ public sealed class NodeManagerHistoryReadPagingTests : IDisposable await host.DisposeAsync(); } + /// Part 11: only a StartTime (EndTime unset = MinValue) with a cap pages FORWARD to an open upper bound — + /// it must not be mistaken for a reverse read, and the stored cursor must not carry MinValue as its end. + [Fact] + public async Task Raw_open_ended_forward_read_pages_ascending_exactly_once() + { + var (host, server) = await BootAsync(); + var nm = server.NodeManager!; + nm.HistoryContinuationStore = new InMemoryHistoryContinuationStore(); + nm.HistorianDataSource = new SeriesHistorianDataSource(MakeSeries(count: 250, stepSeconds: 1)); + nm.EnsureVariable("eq-1/open", parentFolderNodeId: null, displayName: "Open", dataType: "Double", + writable: false, historianTagname: "WW.Open", realm: AddressSpaceRealm.Uns); + var nodeId = nm.TryGetVariable("eq-1/open")!.NodeId; + + var collected = new List(); + byte[]? cp = null; + var pages = 0; + do + { + var (values, error, next) = ReadRaw(nm, nodeId, Epoch, DateTime.MinValue, max: 100, inboundCp: cp); + error.StatusCode.Code.ShouldBe(StatusCodes.Good); + collected.AddRange(values); + cp = next; + (++pages).ShouldBeLessThan(20); + } + while (cp is not null); + + collected.ShouldBe(Enumerable.Range(0, 250).Select(i => (double)i)); + await host.DisposeAsync(); + } + + /// Part 11: only an EndTime (StartTime unset = MinValue) with a cap reads BACKWARD from the EndTime. + [Fact] + public async Task Raw_end_only_read_pages_descending_from_the_end_time() + { + var (host, server) = await BootAsync(); + var nm = server.NodeManager!; + nm.HistoryContinuationStore = new InMemoryHistoryContinuationStore(); + nm.HistorianDataSource = new SeriesHistorianDataSource(MakeSeries(count: 250, stepSeconds: 1)); + nm.EnsureVariable("eq-1/endonly", parentFolderNodeId: null, displayName: "EndOnly", dataType: "Double", + writable: false, historianTagname: "WW.EndOnly", realm: AddressSpaceRealm.Uns); + var nodeId = nm.TryGetVariable("eq-1/endonly")!.NodeId; + + var collected = new List(); + byte[]? cp = null; + var pages = 0; + do + { + var (values, error, next) = ReadRaw(nm, nodeId, DateTime.MinValue, Epoch.AddHours(1), max: 100, inboundCp: cp); + error.StatusCode.Code.ShouldBe(StatusCodes.Good); + collected.AddRange(values); + cp = next; + (++pages).ShouldBeLessThan(20); + } + while (cp is not null); + + collected.ShouldBe(Enumerable.Range(0, 250).Reverse().Select(i => (double)i)); + await host.DisposeAsync(); + } + /// An exactly-full series whose total is a multiple of the cap pages to a trailing /// EMPTY page (GoodNoData, no CP) — the server can't know the prior page was the last until it reads /// past the end.