diff --git a/docs/Historian.md b/docs/Historian.md
index c13e35c8..fbc3808e 100644
--- a/docs/Historian.md
+++ b/docs/Historian.md
@@ -618,6 +618,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
@@ -760,8 +777,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..e9a9a123 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,130 @@ 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!);
+ // 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 };
+ return;
+ }
+
+ sourceName = state.Tagname;
+ startUtc = state.NextStartUtc;
+ endUtc = state.EndUtc;
+ boundarySkip = state.BoundarySkipCount;
+ 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 = 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;
+
+ // 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();
+ }
+
+ // 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))
+ {
+ // 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)
+ {
+ pendingCursor = 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++;
+ 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).
- 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 +5942,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)
+ 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 };
@@ -5867,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.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..6cd34534
--- /dev/null
+++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadEventsPagingTests.cs
@@ -0,0 +1,382 @@
+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();
+ }
+
+ [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) =>
+ 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 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, long resolution = 1)
+ {
+ 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, resolution);
+ nm.HistorianDataSource = src;
+
+ const string equipmentId = "eq-evt";
+ nm.EnsureFolder(equipmentId, parentNodeId: null, displayName: "Equipment", realm: AddressSpaceRealm.Uns);
+ nm.MaterialiseAlarmCondition("alarm-1", equipmentId, "HighTemp", "OffNormalAlarm", severity: 700, realm: AddressSpaceRealm.Uns);
+ return (started.Host, nm, nm.TryGetFolder(equipmentId)!.NodeId, src);
+ }
+
+ private static List 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, end);
+ 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, 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);
+ 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 = end ?? Epoch.AddDays(1),
+ NumValuesPerNode = max,
+ Filter = filter,
+ };
+ var nodesToRead = reads.Select(r => new HistoryReadValueId { NodeId = r.Node, ContinuationPoint = r.Cp }).ToList();
+ var results = reads.Select(_ => (SdkHistoryReadResult)null!).ToList();
+ var errors = reads.Select(_ => (ServiceResult)null!).ToList();
+ nm.HistoryRead(context, details, TimestampsToReturn.Both, false, nodesToRead, results, errors);
+
+ var all = new List<(List, ServiceResult, byte[]?)>();
+ for (var i = 0; i < reads.Length; i++)
+ {
+ 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 all;
+ }
+
+ private static void InvokeRelease(OtOpcUaNodeManager nm, NodeId node, byte[] cp)
+ {
+ var context = new OperationContext(
+ new RequestHeader(), secureChannelContext: null, RequestType.HistoryRead, identity: null);
+ var nodesToRead = new List { 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, 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; }
+
+ public Task ReadEventsAsync(
+ string? sourceName, DateTime startUtc, DateTime endUtc, int maxEvents, CancellationToken ct)
+ {
+ Calls++;
+ // Like the SQL backends: a reversed window is swapped, inputs truncate to the storage resolution.
+ if (startUtc > endUtc) (startUtc, endUtc) = (endUtc, startUtc);
+ startUtc = Trunc(startUtc);
+ endUtc = Trunc(endUtc);
+ var window = All.Where(e => e.EventTimeUtc >= startUtc && e.EventTimeUtc < endUtc).ToList();
+ var cap = maxEvents <= 0 ? 1000 : maxEvents;
+ var truncated = window.Count > cap;
+ return Task.FromResult(new HistoricalEventsResult(
+ window.Take(cap).ToList(), truncated ? new byte[] { 1 } : null));
+ }
+
+ 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 */ }
+ }
+ }
+}
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.