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.
This commit is contained in:
Joseph Doherty
+107
@@ -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;
|
||||
|
||||
/// <summary>
|
||||
/// Gitea #669 over a real SDK client session: an event HistoryRead capped below the available events pages
|
||||
/// through the PRODUCTION session-bound continuation store (<c>SessionHistoryContinuationStore</c>) — the
|
||||
/// path the session-less unit tests cannot reach — and the union of pages is every event exactly once.
|
||||
/// </summary>
|
||||
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<string>();
|
||||
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<string> 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);
|
||||
}
|
||||
|
||||
/// <summary>(time, id) order, [start, end), cap + truncation flag — the SQL backends' contract.</summary>
|
||||
private sealed class SqlLikeEventSource(List<HistoricalEvent> events) : IHistorianDataSource
|
||||
{
|
||||
private readonly List<HistoricalEvent> _all =
|
||||
events.OrderBy(e => e.EventTimeUtc).ThenBy(e => e.EventId, StringComparer.Ordinal).ToList();
|
||||
|
||||
public async Task<HistoricalEventsResult> 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<CoreHistoryReadResult> ReadRawAsync(string r, DateTime s, DateTime e, uint m, CancellationToken ct) =>
|
||||
NullHistorianDataSource.Instance.ReadRawAsync(r, s, e, m, ct);
|
||||
public Task<CoreHistoryReadResult> ReadProcessedAsync(
|
||||
string r, DateTime s, DateTime e, TimeSpan i, HistoryAggregateType a, CancellationToken ct) =>
|
||||
NullHistorianDataSource.Instance.ReadProcessedAsync(r, s, e, i, a, ct);
|
||||
public Task<CoreHistoryReadResult> ReadAtTimeAsync(string r, IReadOnlyList<DateTime> t, CancellationToken ct) =>
|
||||
NullHistorianDataSource.Instance.ReadAtTimeAsync(r, t, ct);
|
||||
public HistorianHealthSnapshot GetHealthSnapshot() => NullHistorianDataSource.Instance.GetHealthSnapshot();
|
||||
public void Dispose() { }
|
||||
}
|
||||
}
|
||||
+258
@@ -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;
|
||||
|
||||
/// <summary>
|
||||
/// Gitea #669 — server-side continuation-point paging for HistoryRead-Events. Same harness as
|
||||
/// <see cref="NodeManagerHistoryReadPagingTests"/>: a real SDK server, a session-less in-process
|
||||
/// HistoryRead and an <see cref="InMemoryHistoryContinuationStore"/>, over a fake source that behaves like
|
||||
/// the SQL backends (ordered by (time, id), half-open window, cap + truncation flag).
|
||||
/// </summary>
|
||||
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<HistoricalEvent>
|
||||
{
|
||||
Ev("e0", 0), Ev("e1", 1), Ev("e2", 2), Ev("e3", 2), Ev("e4", 2), Ev("e5", 3),
|
||||
};
|
||||
var (host, nm, node, _) = await SetupAsync(events);
|
||||
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<HistoricalEvent> { Ev("a0", 0) };
|
||||
for (var i = 0; i < 7; i++) events.Add(Ev($"t{i}", 5));
|
||||
events.Add(Ev("z0", 9));
|
||||
var (host, nm, node, _) = await SetupAsync(events);
|
||||
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<HistoricalEvent>();
|
||||
for (var i = 0; i < 10; i++) events.Add(Ev($"t{i:00}", 5));
|
||||
var (host, nm, node, _) = await SetupAsync(events);
|
||||
nm.MaxTieClusterOverfetch = 4;
|
||||
|
||||
var p1 = ReadEvents(nm, node, max: 3, inboundCp: null);
|
||||
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<HistoricalEvent> 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<HistoricalEvent> 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<OpcUaApplicationHost>.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<string> DrainAsync(OtOpcUaNodeManager nm, NodeId node, uint max)
|
||||
{
|
||||
var all = new List<string>();
|
||||
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<string> 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<HistoryReadValueId> { new() { NodeId = node, ContinuationPoint = inboundCp } };
|
||||
var results = new List<SdkHistoryReadResult> { null! };
|
||||
var errors = new List<ServiceResult> { null! };
|
||||
nm.HistoryRead(context, details, TimestampsToReturn.Both, false, nodesToRead, results, errors);
|
||||
|
||||
var ids = new List<string>();
|
||||
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<HistoryReadValueId> { new() { NodeId = node, ContinuationPoint = cp } };
|
||||
var results = new List<SdkHistoryReadResult> { null! };
|
||||
var errors = new List<ServiceResult> { null! };
|
||||
nm.HistoryRead(context, new ReadEventDetails { StartTime = Epoch, EndTime = Epoch.AddDays(1), NumValuesPerNode = 4 },
|
||||
TimestampsToReturn.Both, true, nodesToRead, results, errors);
|
||||
}
|
||||
|
||||
/// <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
|
||||
{
|
||||
public IReadOnlyList<HistoricalEvent> All { get; } =
|
||||
events.OrderBy(e => e.EventTimeUtc).ThenBy(e => e.EventId, StringComparer.Ordinal).ToList();
|
||||
public int Calls { get; private set; }
|
||||
|
||||
public Task<HistoricalEventsResult> 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<HistorianRead> ReadRawAsync(string r, DateTime s, DateTime e, uint m, CancellationToken ct) =>
|
||||
NullHistorianDataSource.Instance.ReadRawAsync(r, s, e, m, ct);
|
||||
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 HistorianHealthSnapshot GetHealthSnapshot() => NullHistorianDataSource.Instance.GetHealthSnapshot();
|
||||
public void Dispose() { }
|
||||
}
|
||||
|
||||
public void Dispose()
|
||||
{
|
||||
if (Directory.Exists(_pkiRoot))
|
||||
{
|
||||
try { Directory.Delete(_pkiRoot, recursive: true); } catch { /* best effort */ }
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user