feat(historian): #691 HistoryRead-Raw ReturnBounds — Part 11 bounding values on every backend
The node manager composes the start/end bounds from each backend's existing raw reads (one authority, HistoryBounds): at-or-before = a capped reverse read (native, or ReverseRawReader's walk-back on a forward-only backend — never AVEVA's synthetic boundary row), at-or-after = a capped forward read. Bounds count toward NumValuesPerNode, the start bound leads the first page, the end bound ends the last (carried in the continuation point). Missing bound = Bad_BoundNotFound (null, at the bound time; one-sided reads ±1 s from the previous value, Table 1 footnotes a/b). A modified read with ReturnBounds is Bad_InvalidArgument (Part 11 6.5.3.3). The .NET SDK's ReadRawModifiedDetails defaults ReturnBounds to TRUE; the existing tests built details without setting it and now set it false. Oracle: every row of Part 11 Table 1 over three backend shapes, plus paging concatenation, ties, Bad bounds, microsecond truncation; conformance pins the lookup primitives on SqlServer + QuestDB.
This commit is contained in:
@@ -0,0 +1,197 @@
|
||||
using Opc.Ua;
|
||||
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.OpcUaServer;
|
||||
|
||||
/// <summary>
|
||||
/// The end bound still owed to a HistoryRead-Raw read with <c>ReturnBounds = true</c> (Gitea #691). It rides the
|
||||
/// continuation point, because Part 11 returns the end bound with the LAST page only.
|
||||
/// </summary>
|
||||
/// <param name="AtUtc">The request's end time (the later bound of a forward read, the earlier of a reverse one).</param>
|
||||
/// <param name="Reverse">The read is newest-first (<c>StartTime > EndTime</c>, including an end-only read).</param>
|
||||
/// <param name="Inclusive">
|
||||
/// <c>false</c> only for <c>StartTime == EndTime</c>: the samples AT that instant are the data, so the end bound is the
|
||||
/// first sample strictly after it (Part 11 Table 1 footnote c). Otherwise the end bound is the sample at or beyond the
|
||||
/// end time, and the data window excludes the end time.
|
||||
/// </param>
|
||||
/// <param name="Open">The end time was not given (a one-sided read's sentinel): there is no end bound to look up.</param>
|
||||
/// <param name="DataExhausted">The data is drained and the end bound did not fit on the previous page.</param>
|
||||
/// <param name="LastReturnedUtc">The last value's time returned on an earlier page (stamps an open bound's placeholder).</param>
|
||||
internal sealed record HistoryEndBound(
|
||||
DateTime AtUtc, bool Reverse, bool Inclusive, bool Open, bool DataExhausted = false, DateTime? LastReturnedUtc = null);
|
||||
|
||||
/// <summary>
|
||||
/// Gitea #691: OPC UA Part 11 bounding values for HistoryRead-Raw (<c>ReadRawModifiedDetails.ReturnBounds</c>), composed
|
||||
/// in the server from every backend's existing raw read — the ONE authority, so SqlServer, QuestDB, Gateway and Null
|
||||
/// answer identically.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <para>
|
||||
/// <b>What a bound is</b> (Part 11 §4.6, Table 1). Forward: the start bound is the stored sample AT or immediately
|
||||
/// BEFORE StartTime, the end bound the one AT or immediately AFTER EndTime. Reverse mirrors it: the start bound is at
|
||||
/// or after StartTime, the end bound at or before EndTime. A sample exactly on a bound IS the bound: on the start it
|
||||
/// is already the first sample of the data (the window includes StartTime), so nothing extra is added; on the end it is
|
||||
/// added (the window excludes EndTime). <c>StartTime == EndTime</c> reads forward, and its end bound is the first
|
||||
/// sample strictly after the instant. A bound is a stored sample of any quality — a Bad sample is returned as stored.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// <b>Missing bound.</b> <c>Bad_BoundNotFound</c>, a null value, stamped with the bound's time. A one-sided read's
|
||||
/// missing side (no EndTime / no StartTime) is never looked up; it is <c>Bad_BoundNotFound</c> stamped one second after
|
||||
/// (forward) or before (reverse) the previous returned value (Table 1 footnotes a and b).
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// <b>Lookups.</b> "At or before T" is a newest-first read of <c>(0, T]</c> capped at one — native on a backend that
|
||||
/// declares <see cref="IHistorianDataSource.SupportsReverseRaw"/>, otherwise <see cref="ReverseRawReader"/>'s walk-back
|
||||
/// from bounded forward reads (the Gateway: NOT AVEVA's synthetic boundary row, which is stamped at the query start
|
||||
/// rather than at the prior sample's own time). "At or after T" is a forward read of <c>[T, MaxValue)</c> capped at one;
|
||||
/// "strictly after T" starts one storage resolution later (lossless: nothing is stored in between). Inside a tie
|
||||
/// cluster the bound is the cluster's first sample in the read's direction — the forward read's first, the reverse
|
||||
/// read's first (the forward order's last).
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// <b>Resolution.</b> "On the bound" compares at the backend's storage resolution, because a microsecond backend
|
||||
/// (QuestDB) truncates the window bounds it is sent: a sample at <c>⌊T⌋</c> is inside a window starting at <c>T</c>
|
||||
/// there, so it must not also be returned as the start bound.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
internal static class HistoryBounds
|
||||
{
|
||||
private static readonly TimeSpan OpenBoundStep = TimeSpan.FromSeconds(1);
|
||||
|
||||
/// <summary>The end bound owed by a fresh read over the normalised window.</summary>
|
||||
/// <param name="startUtc">The (normalised) start: the later time of a reverse read.</param>
|
||||
/// <param name="endUtc">The (normalised) end; a one-sided read's sentinel (tick 0 / MaxValue) is an open bound.</param>
|
||||
/// <returns>The pending end bound.</returns>
|
||||
public static HistoryEndBound For(DateTime startUtc, DateTime endUtc) =>
|
||||
new(endUtc, Reverse: startUtc > endUtc, Inclusive: startUtc != endUtc, Open: IsOpen(endUtc));
|
||||
|
||||
/// <summary>The pending end bound to store behind a continuation point, remembering the page's last returned time.</summary>
|
||||
/// <param name="bound">The pending end bound (null = bounds not requested).</param>
|
||||
/// <param name="page">The page being returned.</param>
|
||||
/// <returns>The bound to carry, or null.</returns>
|
||||
public static HistoryEndBound? Carry(HistoryEndBound? bound, IReadOnlyList<DataValueSnapshot> page) =>
|
||||
bound is null || page.Count == 0 || page[^1].SourceTimestampUtc is not { } last
|
||||
? bound
|
||||
: bound with { LastReturnedUtc = last };
|
||||
|
||||
/// <summary>The <c>Bad_BoundNotFound</c> placeholder (null value) stamped at <paramref name="atUtc"/>.</summary>
|
||||
/// <param name="atUtc">The bound's time.</param>
|
||||
/// <returns>The placeholder.</returns>
|
||||
public static DataValueSnapshot NotFound(DateTime atUtc) =>
|
||||
new(null, StatusCodes.BadBoundNotFound, atUtc, atUtc);
|
||||
|
||||
/// <summary>
|
||||
/// The start bound to return ahead of the data, or <c>null</c> when a stored sample sits on StartTime (that sample is
|
||||
/// the bound, and it is already the data's first sample).
|
||||
/// </summary>
|
||||
public static async Task<DataValueSnapshot?> StartBoundAsync(
|
||||
IHistorianDataSource source, string tagname, DateTime startUtc, bool reverse, int maxTieCluster, DateTime nowUtc,
|
||||
CancellationToken ct)
|
||||
{
|
||||
if (IsOpen(startUtc)) return NotFound(startUtc);
|
||||
var found = reverse
|
||||
? await EarliestAtOrAfterAsync(source, tagname, startUtc, ct).ConfigureAwait(false)
|
||||
: await LatestAtOrBeforeAsync(source, tagname, startUtc, maxTieCluster, nowUtc, ct).ConfigureAwait(false);
|
||||
if (found is null) return NotFound(startUtc);
|
||||
return SameInstant(Ticks(found), startUtc.Ticks, Resolution(source)) ? null : found;
|
||||
}
|
||||
|
||||
/// <summary>The end bound: the stored sample on or beyond the end time, else the placeholder.</summary>
|
||||
/// <param name="source">The backend.</param>
|
||||
/// <param name="tagname">Historian tag.</param>
|
||||
/// <param name="bound">The pending end bound.</param>
|
||||
/// <param name="previousReturnedUtc">The last value's time already returned (stamps an open bound's placeholder).</param>
|
||||
/// <param name="maxTieCluster">The walk-back's row budget (forward-only backends).</param>
|
||||
/// <param name="nowUtc">Now (the walk-back's look-ahead clamp).</param>
|
||||
/// <param name="ct">Cancellation.</param>
|
||||
public static async Task<DataValueSnapshot> EndBoundAsync(
|
||||
IHistorianDataSource source, string tagname, HistoryEndBound bound, DateTime previousReturnedUtc,
|
||||
int maxTieCluster, DateTime nowUtc, CancellationToken ct)
|
||||
{
|
||||
if (bound.Open)
|
||||
{
|
||||
var ticks = bound.Reverse
|
||||
? Math.Max(0L, previousReturnedUtc.Ticks - OpenBoundStep.Ticks)
|
||||
: Math.Min(DateTime.MaxValue.Ticks, previousReturnedUtc.Ticks + OpenBoundStep.Ticks);
|
||||
return NotFound(new DateTime(ticks, DateTimeKind.Utc));
|
||||
}
|
||||
|
||||
DataValueSnapshot? found;
|
||||
if (bound.Reverse)
|
||||
found = await LatestAtOrBeforeAsync(source, tagname, bound.AtUtc, maxTieCluster, nowUtc, ct).ConfigureAwait(false);
|
||||
else if (bound.Inclusive)
|
||||
found = await EarliestAtOrAfterAsync(source, tagname, bound.AtUtc, ct).ConfigureAwait(false);
|
||||
else
|
||||
{
|
||||
var resolution = Resolution(source);
|
||||
found = bound.AtUtc.Ticks > DateTime.MaxValue.Ticks - resolution
|
||||
? null
|
||||
: await EarliestAtOrAfterAsync(source, tagname, bound.AtUtc.AddTicks(resolution), ct).ConfigureAwait(false);
|
||||
}
|
||||
|
||||
return found ?? NotFound(bound.AtUtc);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Whether a data sample lies inside the data window of a bounded read — a forward window excludes the end time (a
|
||||
/// sample there is the end bound), compared at the storage resolution. A backend whose forward end is inclusive, or a
|
||||
/// resumed read that steps onto the end time, would otherwise return the end bound twice.
|
||||
/// </summary>
|
||||
public static bool InDataWindow(DataValueSnapshot sample, HistoryEndBound bound, long resolution)
|
||||
{
|
||||
if (!bound.Inclusive || bound.Open) return true;
|
||||
var ts = Align(Ticks(sample), resolution);
|
||||
var end = Align(bound.AtUtc.Ticks, resolution);
|
||||
return bound.Reverse ? ts > end : ts < end;
|
||||
}
|
||||
|
||||
/// <summary>Whether a resumed read's next start is already past the data window (only the end bound remains).</summary>
|
||||
public static bool DataWindowDrained(DateTime nextStartUtc, HistoryEndBound bound, long resolution)
|
||||
{
|
||||
if (!bound.Inclusive || bound.Open) return false;
|
||||
var next = Align(nextStartUtc.Ticks, resolution);
|
||||
var end = Align(bound.AtUtc.Ticks, resolution);
|
||||
return bound.Reverse ? next <= end : next >= end;
|
||||
}
|
||||
|
||||
/// <summary>The latest stored sample at or before <paramref name="atUtc"/>, or null.</summary>
|
||||
internal static async Task<DataValueSnapshot?> LatestAtOrBeforeAsync(
|
||||
IHistorianDataSource source, string tagname, DateTime atUtc, int maxTieCluster, DateTime nowUtc, CancellationToken ct)
|
||||
{
|
||||
if (atUtc.Ticks == 0) return null;
|
||||
var floor = new DateTime(0, DateTimeKind.Utc);
|
||||
IReadOnlyList<DataValueSnapshot> rows = source.SupportsReverseRaw
|
||||
? (await source.ReadRawAsync(tagname, atUtc, floor, 1, ct).ConfigureAwait(false)).Samples
|
||||
: (await ReverseRawReader.ReadAsync(source, tagname, atUtc, floor, 1, maxTieCluster, nowUtc, 0, ct)
|
||||
.ConfigureAwait(false)).Samples;
|
||||
var limit = Align(atUtc.Ticks, Resolution(source));
|
||||
foreach (var row in rows)
|
||||
if (Align(Ticks(row), Resolution(source)) <= limit)
|
||||
return row;
|
||||
return null;
|
||||
}
|
||||
|
||||
/// <summary>The earliest stored sample at or after <paramref name="atUtc"/>, or null.</summary>
|
||||
internal static async Task<DataValueSnapshot?> EarliestAtOrAfterAsync(
|
||||
IHistorianDataSource source, string tagname, DateTime atUtc, CancellationToken ct)
|
||||
{
|
||||
if (atUtc.Ticks >= DateTime.MaxValue.Ticks) return null;
|
||||
var rows = (await source.ReadRawAsync(tagname, atUtc, new DateTime(DateTime.MaxValue.Ticks, DateTimeKind.Utc), 1, ct)
|
||||
.ConfigureAwait(false)).Samples;
|
||||
var limit = Align(atUtc.Ticks, Resolution(source));
|
||||
foreach (var row in rows)
|
||||
if (Align(Ticks(row), Resolution(source)) >= limit)
|
||||
return row;
|
||||
return null;
|
||||
}
|
||||
|
||||
private static bool IsOpen(DateTime t) => t.Ticks == 0 || t.Ticks == DateTime.MaxValue.Ticks;
|
||||
|
||||
private static bool SameInstant(long a, long b, long resolution) => Align(a, resolution) == Align(b, resolution);
|
||||
|
||||
private static long Resolution(IHistorianDataSource source) => Math.Max(1L, source.TimestampResolutionTicks);
|
||||
|
||||
private static long Ticks(DataValueSnapshot s) => (s.SourceTimestampUtc ?? DateTime.MinValue).Ticks;
|
||||
|
||||
private static long Align(long ticks, long resolution) => ticks - ticks % resolution;
|
||||
}
|
||||
@@ -62,6 +62,10 @@ namespace ZB.MOM.WW.OtOpcUa.OpcUaServer;
|
||||
/// serialized), so a new field with a "none" default is all the versioning it needs.
|
||||
/// </param>
|
||||
/// <param name="EmptyToTicks">The exclusive upper end of that stretch.</param>
|
||||
/// <param name="PendingEndBound">
|
||||
/// <see cref="HistoryCursorKind.Raw"/> with <c>ReturnBounds = true</c> only (Gitea #691): the end bound still owed —
|
||||
/// Part 11 returns it with the last page. <c>null</c> = bounds not requested.
|
||||
/// </param>
|
||||
internal sealed record HistoryContinuationState(
|
||||
string Tagname,
|
||||
DateTime NextStartUtc,
|
||||
@@ -74,7 +78,8 @@ internal sealed record HistoryContinuationState(
|
||||
bool Reverse = false,
|
||||
long SegmentWidthTicks = 0,
|
||||
long EmptyFromTicks = 0,
|
||||
long EmptyToTicks = 0)
|
||||
long EmptyToTicks = 0,
|
||||
HistoryEndBound? PendingEndBound = null)
|
||||
{
|
||||
/// <summary>The walk-back hint this cursor carries (#686/#689/#691).</summary>
|
||||
public ReverseWalkHint WalkHint => new(SegmentWidthTicks, EmptyFromTicks, EmptyToTicks);
|
||||
|
||||
@@ -5056,6 +5056,15 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
|
||||
var work = new List<Func<Task>>(nodesToProcess.Count);
|
||||
foreach (var handle in nodesToProcess)
|
||||
{
|
||||
if (details.IsReadModified && details.ReturnBounds)
|
||||
{
|
||||
// Part 11 §6.5.3.3: returnBounds "shall be set to FALSE" for a modified read, "otherwise the Server returns
|
||||
// a Bad_InvalidArgument" (#691).
|
||||
errors[handle.Index] = StatusCodes.BadInvalidArgument;
|
||||
results[handle.Index] = new SdkHistoryReadResult { StatusCode = StatusCodes.BadInvalidArgument };
|
||||
continue;
|
||||
}
|
||||
|
||||
if (details.IsReadModified)
|
||||
{
|
||||
// Part 11 modified-value history (#668): the prior values every HistoryUpdate recorded. Gated by the
|
||||
@@ -5070,7 +5079,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
|
||||
work.Add(() => ServeRawPagedAsync(
|
||||
handle, session, nodesToRead, results, errors,
|
||||
details.StartTime, details.EndTime, details.NumValuesPerNode,
|
||||
ct));
|
||||
ct, details.ReturnBounds));
|
||||
}
|
||||
|
||||
RunBounded(work, HistoryReadBatchParallelism);
|
||||
@@ -6197,6 +6206,11 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
|
||||
/// <param name="startTimeUtc">The request window's (inclusive) lower bound, used for a fresh read.</param>
|
||||
/// <param name="endUtc">The (inclusive) upper bound of the read window; unchanged across pages.</param>
|
||||
/// <param name="numValuesPerNode">The client's per-page cap; <c>0</c> means "all values, no paging".</param>
|
||||
/// <param name="ct">Cancellation (the per-request deadline).</param>
|
||||
/// <param name="returnBounds">
|
||||
/// Part 11 bounding values (#691, <see cref="HistoryBounds"/>): the start bound leads the first page, the end bound
|
||||
/// ends the last, and both count toward <paramref name="numValuesPerNode"/>. A resumed read takes it from the cursor.
|
||||
/// </param>
|
||||
private async Task ServeRawPagedAsync(
|
||||
NodeHandle handle,
|
||||
ISession? session,
|
||||
@@ -6206,7 +6220,8 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
|
||||
DateTime startTimeUtc,
|
||||
DateTime endUtc,
|
||||
uint numValuesPerNode,
|
||||
CancellationToken ct)
|
||||
CancellationToken ct,
|
||||
bool returnBounds = false)
|
||||
{
|
||||
var inboundCp = nodesToRead[handle.Index].ContinuationPoint;
|
||||
|
||||
@@ -6215,6 +6230,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
|
||||
DateTime startUtc;
|
||||
var boundarySkip = 0;
|
||||
var walkHint = default(ReverseWalkHint); // #686/#689/#691: the walk-back's width + proven-empty stretch, page to page
|
||||
HistoryEndBound? endBound = null; // #691: the end bound still owed (null = bounds not requested)
|
||||
|
||||
string tagname;
|
||||
if (inboundCp is { Length: > 0 })
|
||||
@@ -6236,6 +6252,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
|
||||
endUtc = state.EndUtc;
|
||||
numValuesPerNode = state.NumValuesPerNode;
|
||||
walkHint = state.WalkHint;
|
||||
endBound = state.PendingEndBound;
|
||||
}
|
||||
else
|
||||
{
|
||||
@@ -6261,6 +6278,78 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
|
||||
else if (startUtc == DateTime.MinValue && endUtc != DateTime.MinValue)
|
||||
(startUtc, endUtc) = (endUtc, new DateTime(0, DateTimeKind.Utc));
|
||||
}
|
||||
|
||||
if (returnBounds) endBound = HistoryBounds.For(startUtc, endUtc);
|
||||
}
|
||||
|
||||
var nowUtc = Clock.GetUtcNow().UtcDateTime;
|
||||
var resolution = Math.Max(1L, HistorianDataSource.TimestampResolutionTicks);
|
||||
|
||||
// #691: the data is drained — the end bound ends this page when it has room (Part 11: bounds count toward
|
||||
// NumValuesPerNode), else it is the whole next page. Returns the continuation point to emit, if any.
|
||||
async Task<byte[]?> AppendEndBoundAsync(List<DataValueSnapshot> page, DateTime resumedAtUtc)
|
||||
{
|
||||
// The previous returned value stamps an open bound's placeholder (Part 11 Table 1 footnotes a/b): this
|
||||
// page's last, else the last of an earlier page (carried in the cursor), else where the read resumed.
|
||||
var previous = (page.Count > 0 ? page[^1].SourceTimestampUtc : null)
|
||||
?? endBound!.LastReturnedUtc ?? resumedAtUtc;
|
||||
if (numValuesPerNode == 0 || page.Count < numValuesPerNode)
|
||||
{
|
||||
page.Add(await HistoryBounds.EndBoundAsync(
|
||||
HistorianDataSource, tagname, endBound!, previous, MaxTieClusterOverfetch, nowUtc, ct).ConfigureAwait(false));
|
||||
return null;
|
||||
}
|
||||
|
||||
return _historyContinuationStore.Save(session, new HistoryContinuationState(
|
||||
tagname, previous, endUtc, 0, numValuesPerNode, SegmentWidthTicks: walkHint.WidthTicks,
|
||||
EmptyFromTicks: walkHint.EmptyFromTicks, EmptyToTicks: walkHint.EmptyToTicks,
|
||||
PendingEndBound: HistoryBounds.Carry(endBound!, page)! with { DataExhausted = true }));
|
||||
}
|
||||
|
||||
// #691: a resumed bounded read whose data is drained (the end bound did not fit on the previous page, or the
|
||||
// resume point already sits on the end time) owes only the end bound.
|
||||
if (endBound is not null && inboundCp is { Length: > 0 }
|
||||
&& (endBound.DataExhausted || HistoryBounds.DataWindowDrained(startUtc, endBound, resolution)))
|
||||
{
|
||||
var onlyBound = new List<DataValueSnapshot>(1);
|
||||
var afterCp = await AppendEndBoundAsync(onlyBound, startUtc).ConfigureAwait(false);
|
||||
results[handle.Index] = new SdkHistoryReadResult
|
||||
{
|
||||
StatusCode = StatusCodes.Good,
|
||||
HistoryData = new ExtensionObject(ToHistoryDataFromSamples(onlyBound)),
|
||||
ContinuationPoint = afterCp,
|
||||
};
|
||||
errors[handle.Index] = ServiceResult.Good;
|
||||
return;
|
||||
}
|
||||
|
||||
// #691: the start bound leads the first page and counts toward the cap; null when a sample sits on StartTime
|
||||
// (that sample is the bound, and it is the data's first).
|
||||
DataValueSnapshot? startBound = null;
|
||||
var readCap = numValuesPerNode;
|
||||
if (endBound is not null && inboundCp is not { Length: > 0 })
|
||||
{
|
||||
startBound = await HistoryBounds.StartBoundAsync(
|
||||
HistorianDataSource, tagname, startUtc, endBound.Reverse, MaxTieClusterOverfetch, nowUtc, ct)
|
||||
.ConfigureAwait(false);
|
||||
if (startBound is not null && numValuesPerNode == 1)
|
||||
{
|
||||
// The page is the start bound alone; the data starts at StartTime on the next page.
|
||||
var boundOnlyCp = _historyContinuationStore.Save(session, new HistoryContinuationState(
|
||||
tagname, startUtc, endUtc, 0, numValuesPerNode, SegmentWidthTicks: walkHint.WidthTicks,
|
||||
EmptyFromTicks: walkHint.EmptyFromTicks, EmptyToTicks: walkHint.EmptyToTicks,
|
||||
PendingEndBound: HistoryBounds.Carry(endBound, [startBound])));
|
||||
results[handle.Index] = new SdkHistoryReadResult
|
||||
{
|
||||
StatusCode = StatusCodes.Good,
|
||||
HistoryData = new ExtensionObject(ToHistoryDataFromSamples([startBound])),
|
||||
ContinuationPoint = boundOnlyCp,
|
||||
};
|
||||
errors[handle.Index] = ServiceResult.Good;
|
||||
return;
|
||||
}
|
||||
|
||||
if (startBound is not null && numValuesPerNode > 0) readCap = numValuesPerNode - 1;
|
||||
}
|
||||
|
||||
// HistoryRead is NOT under the node-manager Lock — awaiting the async source is safe.
|
||||
@@ -6272,8 +6361,8 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
|
||||
if (startUtc > endUtc && !source.SupportsReverseRaw)
|
||||
{
|
||||
var page = await ReverseRawReader.ReadAsync(
|
||||
source, tagname, startUtc, endUtc, numValuesPerNode, MaxTieClusterOverfetch,
|
||||
Clock.GetUtcNow().UtcDateTime, walkHint, ct)
|
||||
source, tagname, startUtc, endUtc, readCap, MaxTieClusterOverfetch,
|
||||
nowUtc, walkHint, ct)
|
||||
.ConfigureAwait(false);
|
||||
walkHint = page.Hint;
|
||||
sourceResult = new HistorianRead(page.Samples, ContinuationPoint: null);
|
||||
@@ -6281,17 +6370,29 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
|
||||
else
|
||||
{
|
||||
sourceResult = await source
|
||||
.ReadRawAsync(tagname, startUtc, endUtc, numValuesPerNode, ct)
|
||||
.ReadRawAsync(tagname, startUtc, endUtc, readCap, ct)
|
||||
.ConfigureAwait(false);
|
||||
}
|
||||
|
||||
var backendFull = HistoryPaging.IsFullPage(sourceResult.Samples.Count, numValuesPerNode);
|
||||
var backendFull = HistoryPaging.IsFullPage(sourceResult.Samples.Count, readCap);
|
||||
|
||||
// On a resume read, drop the boundary ties already returned on the prior page.
|
||||
var samples = inboundCp is { Length: > 0 }
|
||||
? HistoryPaging.TrimBoundaryDuplicates(sourceResult.Samples, startUtc, boundarySkip)
|
||||
: sourceResult.Samples;
|
||||
|
||||
// #691: a bounded read's data window excludes the end time (a sample there is the end bound). A backend whose
|
||||
// forward end is inclusive (the gateway's own) returns it as data; dropping it also means the data is drained.
|
||||
if (endBound is not null)
|
||||
{
|
||||
var inside = samples.Where(x => HistoryBounds.InDataWindow(x, endBound, resolution)).ToList();
|
||||
if (inside.Count != samples.Count)
|
||||
{
|
||||
samples = inside;
|
||||
backendFull = false;
|
||||
}
|
||||
}
|
||||
|
||||
// Oversized tie cluster: a resume read whose FULL backend page does not advance past the resume
|
||||
// boundary timestamp `startUtc`. That happens when more samples share `startUtc` than the page
|
||||
// cap — a tie cluster larger than NumValuesPerNode. The fixed-(start,end,cap) backend can only
|
||||
@@ -6355,9 +6456,15 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
|
||||
{
|
||||
var clusterState = new HistoryContinuationState(
|
||||
tagname, resumeAt, endUtc, nextSkip, numValuesPerNode, SegmentWidthTicks: walkHint.WidthTicks,
|
||||
EmptyFromTicks: walkHint.EmptyFromTicks, EmptyToTicks: walkHint.EmptyToTicks);
|
||||
EmptyFromTicks: walkHint.EmptyFromTicks, EmptyToTicks: walkHint.EmptyToTicks,
|
||||
PendingEndBound: HistoryBounds.Carry(endBound, slice));
|
||||
clusterCp = _historyContinuationStore.Save(session, clusterState);
|
||||
}
|
||||
else if (endBound is not null)
|
||||
{
|
||||
// #691: the data is drained — the end bound ends this page, or is the next one when this one is full.
|
||||
clusterCp = await AppendEndBoundAsync(slice, startUtc).ConfigureAwait(false);
|
||||
}
|
||||
|
||||
results[handle.Index] = new SdkHistoryReadResult
|
||||
{
|
||||
@@ -6371,6 +6478,14 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
|
||||
|
||||
// The "full page" test is against the RAW backend count (before trimming): the backend honoured
|
||||
// the cap, so a full backend page ⇒ there may be more even if we trimmed some boundary ties.
|
||||
// #691: the start bound (first page only) leads the data.
|
||||
if (startBound is not null)
|
||||
{
|
||||
var led = new List<DataValueSnapshot>(samples.Count + 2) { startBound };
|
||||
led.AddRange(samples);
|
||||
samples = led;
|
||||
}
|
||||
|
||||
var historyData = ToHistoryDataFromSamples(samples);
|
||||
|
||||
byte[]? outboundCp = null;
|
||||
@@ -6379,11 +6494,21 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
|
||||
HistoryPaging.ComputeResumeCursor(samples, out var nextStart, out var skip);
|
||||
var nextState = new HistoryContinuationState(
|
||||
tagname, nextStart, endUtc, skip, numValuesPerNode, SegmentWidthTicks: walkHint.WidthTicks,
|
||||
EmptyFromTicks: walkHint.EmptyFromTicks, EmptyToTicks: walkHint.EmptyToTicks);
|
||||
EmptyFromTicks: walkHint.EmptyFromTicks, EmptyToTicks: walkHint.EmptyToTicks,
|
||||
PendingEndBound: HistoryBounds.Carry(endBound, samples));
|
||||
// Save may return null (no session on this request) ⇒ degrade to single-shot for this node.
|
||||
// Built AFTER historyData so a failure projecting samples can never orphan a stored cursor.
|
||||
outboundCp = _historyContinuationStore.Save(session, nextState);
|
||||
}
|
||||
else if (endBound is not null)
|
||||
{
|
||||
// #691: the data is drained — the end bound ends this page, or is the next one when this one is full.
|
||||
var page = new List<DataValueSnapshot>(samples.Count + 1);
|
||||
page.AddRange(samples);
|
||||
outboundCp = await AppendEndBoundAsync(page, startUtc).ConfigureAwait(false);
|
||||
samples = page;
|
||||
historyData = ToHistoryDataFromSamples(samples);
|
||||
}
|
||||
|
||||
results[handle.Index] = new SdkHistoryReadResult
|
||||
{
|
||||
|
||||
Reference in New Issue
Block a user