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;