dohertj2/lmxopcua
1
0
Fork 0
Code Issues 14 Pull Requests Actions Packages Projects Releases Wiki Activity

Merge fix/669-event-continuation: event HistoryRead continuation points + open-ended reads (#669)

Browse Source
This commit is contained in:
Joseph Doherty
2026-10-11 08:37:39 -04:00
parent 91434e7f61 50b504d293
commit d803e0eab4
7 changed files with 719 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/HistoryPaging.cs
+8 -1
View File
@@ -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).
/// </param>
/// <param name="NumValuesPerNode">The client's per-page cap; re-applied to every resumed page.</param>
/// <param name="IsEvents">
/// <c>true</c> for an EVENT HistoryRead cursor (Gitea #669): <paramref name="Tagname"/> then carries the
/// registered historian event-source name and the boundary fields are over <c>EventTimeUtc</c>. 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 <c>BadContinuationPointInvalid</c> miss.
/// </param>
internal sealed record HistoryContinuationState(
string Tagname,
DateTime NextStartUtc,
DateTime EndUtc,
int BoundarySkipCount,
uint NumValuesPerNode);
uint NumValuesPerNode,
bool IsEvents = false);
/// <summary>
/// Pure server-side continuation-point paging decisions for the count-capped variable-history arms
src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs
+128 -17
View File
@@ -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<HistoryReadValueId> nodesToRead,
ISession? session,
IList<SdkHistoryReadResult> results,
IList<ServiceResult> 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<HistoricalEvent> 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<HistoricalEvent>(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.