fix(opcua): event paging review fixes (#669): open-ended end, node-bound cursors, save after projection

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(
+                    pendingCursor = new HistoryContinuationState(
-                        sourceName ?? string.Empty, resumeAt, endUtc, nextSkip, numValuesPerNode, IsEvents: true));
+                        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(
+                pendingCursor = new HistoryContinuationState(
-                    sourceName ?? string.Empty, boundary, endUtc, skip, numValuesPerNode, IsEvents: true));
+                    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 };

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<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);
+        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<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) =>
@@ -131,8 +231,15 @@ public sealed class NodeManagerHistoryReadEventsPagingTests : IDisposable
     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)
+        List<HistoricalEvent> 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<string> DrainAsync(OtOpcUaNodeManager nm, NodeId node, uint max)
+    private static List<string> DrainAsync(OtOpcUaNodeManager nm, NodeId node, uint max, DateTime? end = null)
     {
         var all = new List<string>();
         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<string> 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<string> 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<HistoryReadValueId> { new() { NodeId = node, ContinuationPoint = inboundCp } };
+        var nodesToRead = reads.Select(r => new HistoryReadValueId { NodeId = r.Node, ContinuationPoint = r.Cp }).ToList();
-        var results = new List<SdkHistoryReadResult> { null! };
+        var results = reads.Select(_ => (SdkHistoryReadResult)null!).ToList();
-        var errors = new List<ServiceResult> { null! };
+        var errors = reads.Select(_ => (ServiceResult)null!).ToList();
         nm.HistoryRead(context, details, TimestampsToReturn.Both, false, nodesToRead, results, errors);
 
-        var ids = new List<string>();
+        var all = new List<(List<string>, ServiceResult, byte[]?)>();
-        if (results[0]?.HistoryData is { } ho && !ExtensionObject.IsNull(ho))
+        for (var i = 0; i < reads.Length; i++)
         {
-            var he = (HistoryEvent)ExtensionObject.ToEncodeable(ho);
+            var ids = new List<string>();
-            foreach (var e in he.Events)
+            if (results[i]?.HistoryData is { } ho && !ExtensionObject.IsNull(ho))
-                ids.Add(System.Text.Encoding.UTF8.GetString((byte[])e.EventFields[0].Value));
+            {
+                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
     }
 
     /// <summary>Behaves like the SqlServer/QuestDB event reads: (time, id) order, [start, end), cap, truncation flag.</summary>
-    private sealed class SqlLikeEventSource(List<HistoricalEvent> events) : IHistorianDataSource
+    private sealed class SqlLikeEventSource(List<HistoricalEvent> events, long resolution = 1) : IHistorianDataSource
     {
+        public long TimestampResolutionTicks => resolution;
+        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; }
@@ -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;