feat(historian): #675 reverse HistoryRead-Events paging + end-only backward reads

- IHistorianDataSource.ReadEventsAsync(..., bool reverse, ...) overload: reverse = Part 11 window (lo, hi]
  newest-first by (time, id) DESC. Default forwards forward reads and throws NotSupportedException for
  reverse (Null); Gateway refuses explicitly (ascending-only stream). Server answers
  BadHistoryOperationUnsupported instead of silently returning the earliest N ascending.
- SqlServer / QuestDB serve reverse natively (ORDER BY time DESC, id DESC + TOP/LIMIT).
- Node manager: reverse event reads page with continuation points; direction stored in the cursor
  (HistoryContinuationState.Reverse) so a resume continues in the cursor's direction; reverse tie clusters
  over-fetched forward and sliced from the newest end (SliceTieCluster(reverse:)); end-only reads with a
  cap page backward from EndTime (mirrors raw).
- Resumed event pages over-fetch by the boundary skip, so every non-final page is a full cap page.
- Tests: node-manager reverse paging (ties, clusters, coarse resolution, overfetch bound, resume
  direction both ways, end-only, unsupported backend); shared EventReadConformance on SqlServer +
  QuestDB; over-the-wire reverse paging; offline Gateway pin of the UTC DateTime.MaxValue bound through
  the published client's proto conversion (in-process gRPC capture) and FILETIME.
- docs/Historian.md event paging section.
This commit is contained in:
Joseph Doherty
2026-10-11 11:19:54 -04:00
parent 825393299e
commit 8d154c6a81
20 changed files with 752 additions and 60 deletions
+26 -5
View File
@@ -630,13 +630,31 @@ says is complete carries no continuation point. `NumValuesPerNode == 0` stays un
- 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.
the last event's `EventTimeUtc` **inclusive** and drops the events at that timestamp already emitted — and asks the source for
that many more rows than the cap, so every page short of the last is a FULL `NumValuesPerNode` page (#675; resumed pages used
to come back short by the skip).
- 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.
- **Reverse event reads page newest-first (#675).** `StartTime > EndTime` reads the Part 11 reverse window `(EndTime, StartTime]`
with events in DESCENDING `(EventTimeUtc, EventId)` order, so a cap keeps the **latest** N and a continuation point is issued
when the source flags more. The direction is stored **in the cursor**: a reverse point resumes reverse (and a forward one
forward) whatever the resume request's details say. Reverse tie clusters are over-fetched forward and sliced from the newest
end, exactly as Raw does. Before #675 a capped reverse read silently returned the EARLIEST N, ascending, Good, with no
continuation point.
- **End-only events reads read backward (#675).** `StartTime` unset (`DateTime.MinValue`) with an `EndTime` and
`NumValuesPerNode > 0` pages backward from `EndTime` over `(MinValue, EndTime]` — the same rule as Raw. Start-only reads still
page forward to an open upper bound (`DateTime.MaxValue`, UTC). Without a cap an end-only read keeps its forward
`[MinValue, EndTime)` meaning.
- **Backends.** The server calls `IHistorianDataSource.ReadEventsAsync(…, bool reverse, …)`. SqlServer and QuestDB serve reverse
natively (`… > lo AND … <= hi ORDER BY time DESC, id DESC`). **Gateway and Null refuse a reverse read** (the interface default
throws `NotSupportedException`) and the node answers `BadHistoryOperationUnsupported` — the gateway streams ascending only and
rejects `end < start`, and draining the whole window to keep its tail would be exactly the unbounded read the cap prevents.
- **Gateway open upper bound.** The `DateTime.MaxValue` (UTC) bound of a start-only forward read is pinned offline through the
published client's proto `Timestamp` conversion (captured by an in-process gRPC server) and the FILETIME encoding
(`GatewayEventBoundsTests`). What a live AVEVA Historian does with a year-9999 bound is still unverified (needs the VPN).
- No session on the request (in-process callers) degrades to single-shot, as for Raw.
### Modified-value history (Part 11)
@@ -837,8 +855,10 @@ result never has both.
### Events
`[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
`[start, end)` by `EventTimeUtc`; through the five-argument read **a reversed window (`end < start`) is swapped and returned
ascending**. The direction-aware overload with `reverse: true` (what the node manager calls for a reverse read, #675) reads
`(lo, hi]` ordered `(EventTimeUtc, EventId)` **descending** — the clustered index scanned backwards. A null `sourceName` means all
sources, otherwise `SourceName = @source`. Forward is 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 turns that flag into an OPC UA continuation
point (see [Continuation-point paging (Events)](#continuation-point-paging-events-669)).
@@ -1179,7 +1199,8 @@ Every query filters `NOT deleted`; every value goes through `StoredSample`, so t
from the last timestamp ± `TimestampResolutionTicks`.
- **AtTime:** previous-or-equal and next neighbour of each time by `LIMIT 1` seeks, `UNION ALL`-ed, **50 times per statement** (QuestDB 10.0.1 refuses
more than 126 branches). `AtTimeEvaluator` decides each value, as on `SqlServer`.
- **Events:** `[start, end)` by `ts`, ordered `(ts, event_id)`, `maxEvents + 1` fetched to produce the continuation point.
- **Events:** `[start, end)` by `ts`, ordered `(ts, event_id)`, `maxEvents + 1` fetched to produce the continuation point; a
reverse read (#675) is `(lo, hi]` ordered `(ts, event_id)` descending.
- **Processed:** one query per node; bucket index `(cast(ts AS LONG) * 10 - startUnixTicks) / intervalTicks` (never `SAMPLE BY`, whose alignment
differs), so bucket edges are identical to `SqlServer` (start-inclusive, end-exclusive, partial last bucket, interval 0 ⇒ one bucket, 100,000-bucket cap).
Min/max times use `row_number()` windows with the earliest-timestamp tie-break.
@@ -88,7 +88,7 @@ public sealed class HistoryReadLimitExceededException : InvalidOperationExceptio
}
/// <summary>
/// One row returned by <see cref="IHistorianDataSource.ReadEventsAsync"/> — a historical
/// One row returned by <see cref="IHistorianDataSource.ReadEventsAsync(string?, DateTime, DateTime, int, CancellationToken)"/> — a historical
/// alarm/event record, not the OPC UA live-event stream. Fields match the minimum set the
/// Server needs to populate a <c>HistoryEventFieldList</c> for HistoryReadEvents responses.
/// </summary>
@@ -106,8 +106,8 @@ public sealed record HistoricalEvent(
string? Message,
ushort Severity);
/// <summary>Result of a <see cref="IHistorianDataSource.ReadEventsAsync"/> call.</summary>
/// <param name="Events">Events in chronological order by <c>EventTimeUtc</c>.</param>
/// <summary>Result of a <see cref="IHistorianDataSource.ReadEventsAsync(string?, DateTime, DateTime, int, CancellationToken)"/> call.</summary>
/// <param name="Events">Events ordered by <c>EventTimeUtc</c> in the read's direction — ascending, or descending for a reverse read (#675).</param>
/// <param name="ContinuationPoint">Opaque token for the next call when more events are available; null when complete.</param>
public sealed record HistoricalEventsResult(
IReadOnlyList<HistoricalEvent> Events,
@@ -79,7 +79,7 @@ public interface IHistorianDataSource : IDisposable
/// <c>uint maxValuesPerNode</c> because their OPC UA HistoryRead surface has no
/// equivalent "use default" sentinel.
///
/// This surface declares <see cref="ReadAtTimeAsync"/> and <see cref="ReadEventsAsync"/>
/// This surface declares <see cref="ReadAtTimeAsync"/> and <see cref="ReadEventsAsync(string?, DateTime, DateTime, int, CancellationToken)"/>
/// as required members — a server-side historian owns the full read surface.
/// </remarks>
/// <param name="sourceName">The source name to filter events, or null to return events from all sources.</param>
@@ -101,6 +101,46 @@ public interface IHistorianDataSource : IDisposable
int maxEvents,
CancellationToken cancellationToken);
/// <summary>
/// Read historical alarm / event records in a chosen direction — the overload the server's HistoryReadEvents
/// paging calls (Gitea #675).
/// </summary>
/// <remarks>
/// <para>
/// The bounds may be given in either order; let <c>lo</c>/<c>hi</c> be the earlier/later of them. Forward
/// (<paramref name="reverse"/> <c>false</c>) is exactly <see cref="ReadEventsAsync(string?, DateTime, DateTime, int, CancellationToken)"/>:
/// <c>[lo, hi)</c> ascending by <c>(EventTimeUtc, EventId)</c>. Reverse is the Part 11 reverse window
/// <c>(lo, hi]</c> in DESCENDING <c>(EventTimeUtc, EventId)</c> order, so a cap keeps the LATEST
/// <paramref name="maxEvents"/> events — the mirror of forward, with the same tie-breaker reversed, so a
/// resumed reverse page that re-reads its boundary timestamp sees the boundary ties in the same order.
/// <paramref name="maxEvents"/> and the continuation-point contract are as for the forward read.
/// </para>
/// <para>
/// The default forwards a forward read to the five-argument overload and throws
/// <see cref="NotSupportedException"/> for a reverse one — a backend that cannot read newest-first (Gateway,
/// Null) — which the server reports as <c>BadHistoryOperationUnsupported</c> rather than returning the
/// earliest events in ascending order as if they were the answer.
/// </para>
/// </remarks>
/// <param name="sourceName">The source name to filter events, or null to return events from all sources.</param>
/// <param name="startUtc">One bound of the window (UTC).</param>
/// <param name="endUtc">The other bound of the window (UTC).</param>
/// <param name="maxEvents">The cap; non-positive uses the backend default (see the five-argument overload).</param>
/// <param name="reverse"><c>true</c> to read newest-first over <c>(lo, hi]</c>.</param>
/// <param name="cancellationToken">Cancellation.</param>
/// <returns>The matching events in the read's direction, with a continuation point if truncated.</returns>
/// <exception cref="NotSupportedException">A reverse read on a backend that cannot read newest-first.</exception>
Task<HistoricalEventsResult> ReadEventsAsync(
string? sourceName,
DateTime startUtc,
DateTime endUtc,
int maxEvents,
bool reverse,
CancellationToken cancellationToken) =>
reverse
? throw new NotSupportedException("This historian backend cannot read events newest-first (reverse).")
: ReadEventsAsync(sourceName, startUtc, endUtc, maxEvents, cancellationToken);
/// <summary>
/// Point-in-time health snapshot for diagnostics and dashboards. Pure
/// observation; never blocks on backend I/O.
@@ -125,6 +125,21 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis
}
}
/// <inheritdoc />
/// <remarks>
/// A reverse (newest-first) read is refused with <see cref="NotSupportedException"/> — the server reports
/// <c>BadHistoryOperationUnsupported</c> (Gitea #675). The gateway's <c>ReadEvents</c> streams ascending only and
/// rejects <c>end &lt; start</c>, so the only way to serve "the latest N" would be to drain the whole window and
/// keep its tail, an unbounded read the cap exists to prevent. Nothing is sent.
/// </remarks>
public Task<HistoricalEventsResult> ReadEventsAsync(
string? sourceName, DateTime startUtc, DateTime endUtc, int maxEvents, bool reverse,
CancellationToken cancellationToken) =>
reverse
? throw new NotSupportedException(
"The HistorianGateway backend reads events oldest-first only; a reverse (newest-first) events read is not supported.")
: ReadEventsAsync(sourceName, startUtc, endUtc, maxEvents, cancellationToken);
/// <inheritdoc />
public async Task<HistoricalEventsResult> ReadEventsAsync(
string? sourceName, DateTime startUtc, DateTime endUtc, int maxEvents,
@@ -361,8 +361,18 @@ public sealed class QuestDbHistorianDataSource : IHistorianDataSource
/// <c>maxEvents + 1</c> rows are fetched; when the extra row arrives it is dropped and the continuation point is
/// the UTF-8 of <see cref="EventContinuation.Encode"/> for that first not-returned event.
/// </remarks>
public Task<HistoricalEventsResult> ReadEventsAsync(
string? sourceName, DateTime startUtc, DateTime endUtc, int maxEvents, CancellationToken cancellationToken) =>
ReadEventsAsync(sourceName, startUtc, endUtc, maxEvents, reverse: false, cancellationToken);
/// <inheritdoc />
/// <remarks>
/// Forward is the five-argument read. Reverse (Gitea #675) is the (rounded) <c>(lo, hi]</c> ordered by
/// <c>(ts, event_id)</c> DESCENDING, so <c>LIMIT</c> keeps the latest events; truncation and the continuation point
/// work as forward.
/// </remarks>
public async Task<HistoricalEventsResult> ReadEventsAsync(
string? sourceName, DateTime startUtc, DateTime endUtc, int maxEvents, CancellationToken cancellationToken)
string? sourceName, DateTime startUtc, DateTime endUtc, int maxEvents, bool reverse, CancellationToken cancellationToken)
{
// No stored event can carry a source longer than the limit, so an over-long filter matches nothing.
if (sourceName is not null && sourceName.Length > HistorianTagNames.MaxLength)
@@ -376,9 +386,9 @@ public sealed class QuestDbHistorianDataSource : IHistorianDataSource
var bounded = maxEvents > 0 && maxEvents < int.MaxValue;
var sql =
$"SELECT event_id, source, ts, received_utc, message, severity FROM {_conn.EventTable} " +
"WHERE ts >= $1 AND ts < $2" +
(reverse ? "WHERE ts > $1 AND ts <= $2" : "WHERE ts >= $1 AND ts < $2") +
(sourceName is null ? "" : " AND source = $3") +
" ORDER BY ts ASC, event_id ASC" +
(reverse ? " ORDER BY ts DESC, event_id DESC" : " ORDER BY ts ASC, event_id ASC") +
(bounded ? (sourceName is null ? " LIMIT $3" : " LIMIT $4") : "");
await using var conn = await _conn.OpenAsync(cancellationToken).ConfigureAwait(false);
@@ -314,8 +314,18 @@ public sealed class SqlServerHistorianDataSource : IHistorianDataSource
/// of <c>"{EventTimeUtc.Ticks}|{EventId}"</c> of that first not-returned event (opaque to callers; only
/// its presence is contractual).
/// </remarks>
public Task<HistoricalEventsResult> ReadEventsAsync(
string? sourceName, DateTime startUtc, DateTime endUtc, int maxEvents, CancellationToken cancellationToken) =>
ReadEventsAsync(sourceName, startUtc, endUtc, maxEvents, reverse: false, cancellationToken);
/// <inheritdoc />
/// <remarks>
/// Forward is the five-argument read. Reverse (Gitea #675) is <c>(lo, hi]</c> ordered by
/// <c>(EventTimeUtc, EventId)</c> DESCENDING — the clustered index scanned backwards — so <c>TOP</c> keeps the latest
/// events; truncation and the continuation point work as forward.
/// </remarks>
public async Task<HistoricalEventsResult> ReadEventsAsync(
string? sourceName, DateTime startUtc, DateTime endUtc, int maxEvents, CancellationToken cancellationToken)
string? sourceName, DateTime startUtc, DateTime endUtc, int maxEvents, bool reverse, CancellationToken cancellationToken)
{
// No stored event can carry a source longer than the column (the alarm writer dead-letters it rather than
// truncating), so an over-long filter matches nothing — never a truncated prefix of someone else's source.
@@ -328,9 +338,10 @@ public sealed class SqlServerHistorianDataSource : IHistorianDataSource
var bounded = maxEvents > 0 && maxEvents < int.MaxValue;
var sql =
$"SELECT {(bounded ? "TOP (@take) " : "")}EventId, SourceName, EventTimeUtc, ReceivedTimeUtc, Message, Severity " +
$"FROM {_conn.Q("Event")} WHERE EventTimeUtc >= @lo AND EventTimeUtc < @hi" +
$"FROM {_conn.Q("Event")} WHERE " +
(reverse ? "EventTimeUtc > @lo AND EventTimeUtc <= @hi" : "EventTimeUtc >= @lo AND EventTimeUtc < @hi") +
(sourceName is null ? "" : " AND SourceName = @source") +
" ORDER BY EventTimeUtc ASC, EventId ASC";
(reverse ? " ORDER BY EventTimeUtc DESC, EventId DESC" : " ORDER BY EventTimeUtc ASC, EventId ASC");
await using var conn = await _conn.OpenAsync(cancellationToken).ConfigureAwait(false);
await using var cmd = _conn.CreateCommand(conn, sql);
@@ -46,6 +46,11 @@ namespace ZB.MOM.WW.OtOpcUa.OpcUaServer;
/// <see cref="HistoryCursorKind.Modified"/> only: the normalised window start, so a resumed page keeps the original
/// window and direction (the modified read resumes by key, not by moving the window start).
/// </param>
/// <param name="Reverse">
/// <see cref="HistoryCursorKind.Events"/> only (Gitea #675): the read is newest-first over
/// <c>(EndUtc, NextStartUtc]</c>. A resumed events read takes its direction from here, never from the resume
/// request's details. Raw keeps deriving it from the window (<c>NextStartUtc &gt; EndUtc</c>).
/// </param>
internal sealed record HistoryContinuationState(
string Tagname,
DateTime NextStartUtc,
@@ -54,7 +59,8 @@ internal sealed record HistoryContinuationState(
uint NumValuesPerNode,
HistoryCursorKind Kind = HistoryCursorKind.Raw,
long LastModifiedId = 0,
DateTime WindowStartUtc = default);
DateTime WindowStartUtc = default,
bool Reverse = false);
/// <summary>The HistoryRead kind that issued a <see cref="HistoryContinuationState"/>.</summary>
internal enum HistoryCursorKind
@@ -102,7 +102,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
/// <see cref="_alarmConditions"/> / <see cref="_notifierFolders"/>.</summary>
private readonly Dictionary<RealmKey, AlarmNotifierWiring> _alarmNotifierWiring = new();
/// <summary>Phase C: event-notifier folder NodeId-identifier → the event-history source
/// name passed to <see cref="IHistorianDataSource.ReadEventsAsync"/>. The equipment-folder NodeId
/// name passed to <see cref="IHistorianDataSource.ReadEventsAsync(string?, DateTime, DateTime, int, CancellationToken)"/>. The equipment-folder NodeId
/// identifier IS the equipment id, which IS the sourceName, so key and value are the same string;
/// the map's presence (not its value) is what makes a folder an event-history source. Populated by
/// <see cref="EnsureFolderIsEventNotifier"/> only when a real historian is wired at promotion time,
@@ -5153,6 +5153,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
var endUtc = details.EndTime;
var numValuesPerNode = details.NumValuesPerNode;
var boundarySkip = 0;
bool reverse;
if (resumed)
{
@@ -5172,36 +5173,59 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
endUtc = state.EndUtc;
boundarySkip = state.BoundarySkipCount;
numValuesPerNode = state.NumValuesPerNode;
// The direction is the CURSOR's (Gitea #675), never re-derived from the resume request's details: a
// reverse read resumes reverse, a forward one forward.
reverse = state.Reverse;
}
else
{
// Part 11: with one bound unset, 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 the read would
// look reversed. An unset StartTime with an EndTime and a count cap means "page BACKWARD from EndTime"
// (#675, mirroring raw): the reverse window (MinValue, EndTime]. Without a count cap an end-only read
// keeps its historical forward [MinValue, EndTime) meaning.
if (endUtc == DateTime.MinValue && startUtc != DateTime.MinValue)
endUtc = new DateTime(DateTime.MaxValue.Ticks, DateTimeKind.Utc); // Utc kind: a proto Timestamp conversion rejects Unspecified
else if (numValuesPerNode > 0 && startUtc == DateTime.MinValue && endUtc != DateTime.MinValue)
(startUtc, endUtc) = (endUtc, new DateTime(0, DateTimeKind.Utc));
// 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;
// Part 11: StartTime later than EndTime is a reverse read — events newest-first over (end, start].
reverse = 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.
// to UNBOUNDED (EventMaxEvents) rather than the backend's <=0 default-cap sentinel. A resumed page re-reads its
// boundary timestamp inclusively and then drops the `boundarySkip` ties already emitted, so it asks for that
// many more rows than the cap — otherwise every resumed page would come back short by the skip.
var fetch = EventMaxEvents(numValuesPerNode);
if (numValuesPerNode > 0 && boundarySkip > 0)
fetch = (int)Math.Min(int.MaxValue - 1L, (long)fetch + boundarySkip);
var sourceResult = await _historianDataSource.ReadEventsAsync(
sourceName, startUtc, endUtc, EventMaxEvents(numValuesPerNode), ct).ConfigureAwait(false);
sourceName, startUtc, endUtc, fetch, reverse, 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;
var truncated = sourceResult.ContinuationPoint is not null && numValuesPerNode > 0;
IReadOnlyList<HistoricalEvent> events = sourceResult.Events;
if (resumed && boundarySkip > 0)
{
// Drop the events at the boundary timestamp already emitted on the prior page.
// Drop the events at the boundary timestamp already emitted on the prior page. The resumed read is in
// the cursor's direction, so in both directions those ties lead the page.
var i = 0;
while (i < events.Count && i < boundarySkip && events[i].EventTimeUtc == startUtc) i++;
if (i > 0) events = events.Skip(i).ToList();
}
// The skip over-fetch can leave more than the cap (fewer boundary ties came back than were skipped — pruned
// between pages): keep the cap, and the dropped tail means the read continues.
if (numValuesPerNode > 0 && events.Count > numValuesPerNode)
{
events = events.Take((int)Math.Min(int.MaxValue, numValuesPerNode)).ToList();
truncated = true;
}
// 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;
@@ -5214,10 +5238,14 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
// 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)
IReadOnlyList<HistoricalEvent> cluster = (await _historianDataSource.ReadEventsAsync(
sourceName, startUtc, startUtc.AddTicks(resolution), overfetchCap, reverse: false, ct)
.ConfigureAwait(false)).Events;
// The over-fetch reads the cluster FORWARD (ascending EventId), but a reverse read emitted its ties — and
// counted boundarySkip — in descending order. Slice in the read's own direction, as raw does.
if (reverse) cluster = cluster.Reverse().ToList();
if (cluster.Count > MaxTieClusterOverfetch)
{
_logger.LogError(
@@ -5232,7 +5260,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
HistoryPaging.SliceTieCluster(
cluster.Count, boundarySkip, numValuesPerNode, startUtc, endUtc,
out var sliceStart, out var sliceCount, out var nextStartUtc, out var nextSkip,
reverse: false, resolutionTicks: resolution);
reverse: reverse, resolutionTicks: resolution);
var slice = new List<HistoricalEvent>(sliceCount);
for (var i = sliceStart; i < sliceStart + sliceCount; i++) slice.Add(cluster[i]);
@@ -5241,17 +5269,20 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
if (nextStartUtc is { } resumeAt)
{
pendingCursor = new HistoryContinuationState(
sourceName ?? string.Empty, resumeAt, endUtc, nextSkip, numValuesPerNode, HistoryCursorKind.Events);
sourceName ?? string.Empty, resumeAt, endUtc, nextSkip, numValuesPerNode, HistoryCursorKind.Events,
Reverse: reverse);
}
}
else if (truncated && events.Count > 0)
{
// Cursor from the last event: resume at its time INCLUSIVE, skipping the trailing ties emitted.
// Cursor from the last event: resume at its time INCLUSIVE, skipping the trailing ties emitted. In a
// reverse read the last event is the OLDEST on the page, and the resumed (end, T] read starts there.
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, HistoryCursorKind.Events);
sourceName ?? string.Empty, boundary, endUtc, skip, numValuesPerNode, HistoryCursorKind.Events,
Reverse: reverse);
}
var historyEvent = ProjectEvents(events, selectClauses);
@@ -5272,6 +5303,15 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
errors[handle.Index] = StatusCodes.BadTimeout;
results[handle.Index] = new SdkHistoryReadResult { StatusCode = StatusCodes.BadTimeout };
}
catch (NotSupportedException ex)
{
// The backend cannot serve this read's shape — today a reverse (newest-first) read on Gateway / Null
// (#675). A client request the backend refuses, not a backend fault: Warning, not Error.
_logger.LogWarning("OtOpcUaNodeManager: HistoryReadEvents for node {NodeId} is not supported by the historian backend: {Reason}",
handle.NodeId, ex.Message);
errors[handle.Index] = StatusCodes.BadHistoryOperationUnsupported;
results[handle.Index] = new SdkHistoryReadResult { StatusCode = StatusCodes.BadHistoryOperationUnsupported };
}
catch (Exception ex)
{
// One node's backend failure must not throw out of the batch — surface Bad for THIS node
@@ -5611,7 +5651,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
private static int ClampToInt(uint value) => value > int.MaxValue ? int.MaxValue : (int)value;
/// <summary>Map a HistoryRead-Events <c>NumValuesPerNode</c> request cap onto the
/// <see cref="IHistorianDataSource.ReadEventsAsync"/> <c>maxEvents</c> argument, honouring the OPC UA
/// <see cref="IHistorianDataSource.ReadEventsAsync(string?, DateTime, DateTime, int, CancellationToken)"/> <c>maxEvents</c> argument, honouring the OPC UA
/// Part 4/11 semantics that <c>NumValuesPerNode == 0</c> means "no limit — return ALL values".
/// We translate 0 to UNBOUNDED (<see cref="int.MaxValue"/>) — a very large positive cap — rather than
/// the backend's <c>maxEvents &lt;= 0</c> "use the default cap" sentinel, so a client asking for the
@@ -45,9 +45,10 @@ public sealed class IHistorianDataSourceContractTests
[InlineData("ReadEventsAsync", typeof(Task<HistoricalEventsResult>))]
public void ReadMethods_ReturnExpectedTaskShape(string methodName, Type expectedReturnType)
{
var method = typeof(IHistorianDataSource).GetMethod(methodName);
method.ShouldNotBeNull();
method!.ReturnType.ShouldBe(expectedReturnType);
// ReadEventsAsync is overloaded (the #675 direction-aware overload) — every overload must keep the shape.
var methods = Overloads(methodName);
methods.ShouldNotBeEmpty();
foreach (var method in methods) method.ReturnType.ShouldBe(expectedReturnType);
}
/// <summary>Verifies that GetHealthSnapshot is synchronous.</summary>
@@ -146,11 +147,14 @@ public sealed class IHistorianDataSourceContractTests
[InlineData("ReadEventsAsync", "maxEvents", typeof(int))]
public void HistoryRead_MaxParameter_TypePinned(string methodName, string parameterName, Type expectedType)
{
var method = typeof(IHistorianDataSource).GetMethod(methodName);
method.ShouldNotBeNull();
var parameter = method!.GetParameters().FirstOrDefault(p => p.Name == parameterName);
parameter.ShouldNotBeNull($"Method {methodName} should expose a '{parameterName}' parameter.");
parameter!.ParameterType.ShouldBe(expectedType);
var methods = Overloads(methodName);
methods.ShouldNotBeEmpty();
foreach (var method in methods)
{
var parameter = method.GetParameters().FirstOrDefault(p => p.Name == parameterName);
parameter.ShouldNotBeNull($"Method {methodName} should expose a '{parameterName}' parameter.");
parameter!.ParameterType.ShouldBe(expectedType);
}
}
/// <summary>
@@ -163,13 +167,17 @@ public sealed class IHistorianDataSourceContractTests
[Fact]
public void ReadEventsAsync_sentinel_and_continuation_contract_types_pinned()
{
var method = typeof(IHistorianDataSource).GetMethod("ReadEventsAsync");
method.ShouldNotBeNull();
var methods = Overloads("ReadEventsAsync");
methods.Count.ShouldBe(2, "the five-argument read plus the #675 direction-aware overload");
methods.ShouldContain(m => m.GetParameters().Any(p => p.Name == "reverse" && p.ParameterType == typeof(bool)));
var maxEventsParam = method!.GetParameters().FirstOrDefault(p => p.Name == "maxEvents");
maxEventsParam.ShouldNotBeNull("maxEvents parameter must exist for the sentinel contract");
maxEventsParam!.ParameterType.ShouldBe(typeof(int),
"maxEvents is int (not uint) so callers can pass <=0 as a 'use backend default cap' sentinel");
foreach (var method in methods)
{
var maxEventsParam = method.GetParameters().FirstOrDefault(p => p.Name == "maxEvents");
maxEventsParam.ShouldNotBeNull("maxEvents parameter must exist for the sentinel contract");
maxEventsParam!.ParameterType.ShouldBe(typeof(int),
"maxEvents is int (not uint) so callers can pass <=0 as a 'use backend default cap' sentinel");
}
// HistoricalEventsResult must carry a nullable ContinuationPoint so implementations
// can signal 'more results exist' (non-null) vs 'all results returned' (null).
@@ -179,4 +187,7 @@ public sealed class IHistorianDataSourceContractTests
continuationProp!.PropertyType.ShouldBe(typeof(byte[]),
"ContinuationPoint is byte[]? — null means no more pages, non-null means caller must page");
}
private static List<System.Reflection.MethodInfo> Overloads(string name) =>
typeof(IHistorianDataSource).GetMethods().Where(m => m.Name == name).ToList();
}
@@ -67,6 +67,20 @@ public sealed class NullHistorianDataSourceTests
result.ContinuationPoint.ShouldBeNull();
}
/// <summary>
/// #675: the direction-aware overload's default — forward is the five-argument read, reverse is refused
/// (the server answers <c>BadHistoryOperationUnsupported</c>).
/// </summary>
[Fact]
public async Task Direction_aware_ReadEventsAsync_forwards_forward_and_refuses_reverse()
{
IHistorianDataSource source = Source;
(await source.ReadEventsAsync(null, DateTime.UtcNow.AddHours(-1), DateTime.UtcNow, 0, reverse: false, Ct))
.Events.ShouldBeEmpty();
await Should.ThrowAsync<NotSupportedException>(() =>
source.ReadEventsAsync(null, DateTime.UtcNow, DateTime.UtcNow.AddHours(-1), 0, reverse: true, Ct));
}
/// <summary>GetHealthSnapshot reports a fully-disabled (zeroed, disconnected, no-node) shape.</summary>
[Fact]
public void GetHealthSnapshot_is_disabled()
@@ -0,0 +1,106 @@
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Conformance;
/// <summary>
/// The direction-aware event read (Gitea #675): forward is <c>[lo, hi)</c> ascending by <c>(EventTimeUtc, EventId)</c>;
/// reverse is the Part 11 reverse window <c>(lo, hi]</c> DESCENDING by the same key, so a capped reverse read returns the
/// LATEST events and flags truncation. Paging a reverse read the way the server does — resume at the last returned
/// time inclusive and drop the boundary ties already emitted — must visit every event exactly once, newest-first,
/// through a same-timestamp tie cluster.
/// </summary>
public abstract class EventReadConformance : ConformanceBase
{
private static readonly CancellationToken Ct = CancellationToken.None;
/// <summary>e0@0, e1@1, then the ties t0..t4@2, then e5@3 and e6@4 (all source "s"), plus o0@2 from another source.</summary>
private async Task<IHistorianDataSource> SeedAsync()
{
await B.SeedEventsAsync(
(At(0), "e0", "s"), (At(1), "e1", "s"),
(At(2), "t0", "s"), (At(2), "t1", "s"), (At(2), "t2", "s"), (At(2), "t3", "s"), (At(2), "t4", "s"),
(At(3), "e5", "s"), (At(4), "e6", "s"),
(At(2), "o0", "other"));
return B.CreateDataSource();
}
private static readonly string[] NewestFirst = ["e6", "e5", "t4", "t3", "t2", "t1", "t0", "e1", "e0"];
[Fact]
public async Task Reverse_read_is_newest_first_by_time_then_EventId_descending()
{
using var src = await SeedAsync();
var r = await src.ReadEventsAsync("s", At(-1), At(10), 0, reverse: true, Ct);
r.Events.Select(e => e.EventId).ShouldBe(NewestFirst);
r.ContinuationPoint.ShouldBeNull();
}
[Fact]
public async Task Reverse_window_is_end_exclusive_start_inclusive_whichever_order_the_bounds_come_in()
{
using var src = await SeedAsync();
// (At(1), At(3)]: e1 at the lower bound is excluded, e5 at the upper bound included.
(await src.ReadEventsAsync("s", At(3), At(1), 0, reverse: true, Ct)).Events.Select(e => e.EventId)
.ShouldBe(["e5", "t4", "t3", "t2", "t1", "t0"]);
(await src.ReadEventsAsync("s", At(1), At(3), 0, reverse: true, Ct)).Events.Select(e => e.EventId)
.ShouldBe(["e5", "t4", "t3", "t2", "t1", "t0"]);
// Forward is unchanged: [At(1), At(3)) ascending.
(await src.ReadEventsAsync("s", At(1), At(3), 0, reverse: false, Ct)).Events.Select(e => e.EventId)
.ShouldBe(["e1", "t0", "t1", "t2", "t3", "t4"]);
}
[Fact]
public async Task A_capped_reverse_read_keeps_the_LATEST_events_and_flags_truncation()
{
using var src = await SeedAsync();
var r = await src.ReadEventsAsync("s", At(10), At(-1), 3, reverse: true, Ct);
r.Events.Select(e => e.EventId).ShouldBe(["e6", "e5", "t4"]);
r.ContinuationPoint.ShouldNotBeNull();
// An exact fit is complete.
(await src.ReadEventsAsync("s", At(10), At(-1), 9, reverse: true, Ct)).ContinuationPoint.ShouldBeNull();
}
[Fact]
public async Task Paging_a_reverse_read_through_a_tie_cluster_visits_every_event_once_newest_first()
{
using var src = await SeedAsync();
foreach (var cap in new[] { 1, 2, 3, 4 })
{
var all = new List<string>();
var lo = At(-1);
var from = At(10);
var skip = 0;
for (var page = 0; ; page++)
{
page.ShouldBeLessThan(40, "paging must terminate");
var r = await src.ReadEventsAsync("s", from, lo, cap + skip, reverse: true, Ct);
var events = r.Events.ToList();
var drop = 0;
while (drop < skip && drop < events.Count && events[drop].EventTimeUtc == from) drop++;
events = events.Skip(drop).ToList();
// The server asks for cap + skip rows (it re-reads the boundary ties it then drops), so a page is never
// emptied by the trim while more remains.
if (r.ContinuationPoint is not null) events.Count.ShouldBe(cap);
all.AddRange(events.Select(e => e.EventId));
if (r.ContinuationPoint is null) break;
var boundary = events[^1].EventTimeUtc;
var ties = events.AsEnumerable().Reverse().TakeWhile(e => e.EventTimeUtc == boundary).Count();
skip = boundary == from ? skip + ties : ties;
from = boundary;
}
all.SequenceEqual(NewestFirst).ShouldBeTrue($"cap {cap}: {string.Join(",", all)}");
}
}
[Fact]
public async Task Reverse_read_with_a_null_source_spans_every_source()
{
using var src = await SeedAsync();
var r = await src.ReadEventsAsync(null, At(2.5), At(1.5), 0, reverse: true, Ct);
r.Events.Select(e => e.EventId).ShouldBe(["t4", "t3", "t2", "t1", "t0", "o0"]);
}
}
@@ -41,6 +41,13 @@ public interface IConformanceBackend : IAsyncDisposable
/// <param name="samples">The rows to store.</param>
Task SeedAsync(string tag, params (DateTime T, HistorianValue V, uint Status)[] samples);
/// <summary>
/// Writes historical events directly in the backend's event table (never through the alarm writer): event id,
/// source and time as given (rounded to the storage resolution), severity 500, received time = event time.
/// </summary>
/// <param name="events">The events to store.</param>
Task SeedEventsAsync(params (DateTime T, string Id, string Source)[] events);
/// <summary>
/// Live (non-deleted) rows for a tag, ascending by time, read directly from storage. The status is returned
/// VERBATIM as stored — there is no read-side clamp (an Empty row stored non-Bad reads back non-Bad here).
@@ -0,0 +1,123 @@
using System.Net;
using Google.Protobuf.WellKnownTypes;
using Grpc.Core;
using Microsoft.AspNetCore.Builder;
using Microsoft.AspNetCore.Hosting;
using Microsoft.AspNetCore.Server.Kestrel.Core;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Logging.Abstractions;
using Shouldly;
using Xunit;
using ZB.MOM.WW.HistorianGateway.Contracts.Grpc;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Runtime.Historian;
namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests;
/// <summary>
/// Gitea #675: the Gateway backend's event-read bounds and direction, offline.
/// <list type="bullet">
/// <item>A reverse (newest-first) read is refused before anything is sent — the gateway streams ascending only.</item>
/// <item>The open upper bound a start-only forward read uses (<c>DateTime.MaxValue</c>, UTC kind) survives the
/// PUBLISHED client package's proto conversion — captured on the wire by an in-process gRPC server — and the
/// Windows FILETIME encoding the gateway converts it to. The live AVEVA leg (what the historian does with a
/// year-9999 bound) stays open: it needs the VPN.</item>
/// </list>
/// </summary>
public sealed class GatewayEventBoundsTests
{
private static readonly DateTime UtcMax = new(DateTime.MaxValue.Ticks, DateTimeKind.Utc);
private static CancellationToken Ct => TestContext.Current.CancellationToken;
[Fact]
public async Task A_reverse_events_read_is_refused_and_sends_nothing()
{
var fake = new FakeHistorianGatewayClient();
IHistorianDataSource ds = new GatewayHistorianDataSource(fake, NullLogger<GatewayHistorianDataSource>.Instance);
await Should.ThrowAsync<NotSupportedException>(() =>
ds.ReadEventsAsync("Pump1", DateTime.UtcNow, DateTime.UtcNow.AddHours(-1), 10, reverse: true, Ct));
fake.ReadEventsCallCount.ShouldBe(0);
// Forward through the same overload is the five-argument read, unchanged.
await ds.ReadEventsAsync("Pump1", DateTime.UtcNow.AddHours(-1), DateTime.UtcNow, 10, reverse: false, Ct);
fake.ReadEventsCallCount.ShouldBe(1);
}
[Fact]
public void UtcMax_round_trips_through_the_proto_Timestamp_and_FILETIME_encodings()
{
var ts = Timestamp.FromDateTime(UtcMax);
ts.ToDateTime().ShouldBe(UtcMax);
ts.ToDateTime().Kind.ShouldBe(DateTimeKind.Utc);
// The gateway hands its bounds to the historian as FILETIME (100 ns since 1601) — in range up to year 30828.
var fileTime = ts.ToDateTime().ToFileTimeUtc();
DateTime.FromFileTimeUtc(fileTime).ShouldBe(UtcMax);
}
[Fact]
public async Task A_start_only_bound_of_UtcMax_reaches_the_wire_intact_through_the_published_client()
{
var capture = new CapturingReadService();
await using var server = await StartServerAsync(capture);
using var adapter = HistorianGatewayClientAdapter.Create(
new ServerHistorianOptions
{
Enabled = true,
Endpoint = $"http://127.0.0.1:{server.Port}",
ApiKey = "histgw_test_key",
UseTls = false,
},
NullLoggerFactory.Instance);
IHistorianDataSource ds = new GatewayHistorianDataSource(adapter, NullLogger<GatewayHistorianDataSource>.Instance);
var start = new DateTime(2026, 1, 1, 0, 0, 0, DateTimeKind.Utc);
var r = await ds.ReadEventsAsync("Pump1", start, UtcMax, 5, Ct);
r.Events.ShouldBeEmpty();
capture.Request.ShouldNotBeNull();
capture.Request!.StartTime.ToDateTime().ShouldBe(start);
capture.Request.EndTime.ToDateTime().ShouldBe(UtcMax);
DateTime.FromFileTimeUtc(capture.Request.EndTime.ToDateTime().ToFileTimeUtc()).ShouldBe(UtcMax);
}
private sealed class CapturingReadService : HistorianRead.HistorianReadBase
{
public ReadEventsRequest? Request { get; private set; }
public override Task ReadEvents(
ReadEventsRequest request, IServerStreamWriter<HistorianEvent> responseStream, ServerCallContext context)
{
Request = request;
return Task.CompletedTask;
}
}
private sealed class Server(WebApplication app, int port) : IAsyncDisposable
{
public int Port { get; } = port;
public async ValueTask DisposeAsync()
{
await app.StopAsync();
await app.DisposeAsync();
}
}
private static async Task<Server> StartServerAsync(CapturingReadService service)
{
var builder = WebApplication.CreateSlimBuilder();
builder.Logging.ClearProviders();
builder.WebHost.ConfigureKestrel(k =>
k.Listen(IPAddress.Loopback, 0, o => o.Protocols = HttpProtocols.Http2));
builder.Services.AddGrpc();
builder.Services.AddSingleton(service);
var app = builder.Build();
app.MapGrpcService<CapturingReadService>();
await app.StartAsync();
var port = new Uri(app.Urls.First()).Port;
return new Server(app, port);
}
}
@@ -15,12 +15,18 @@
<PackageReference Include="Microsoft.NET.Test.Sdk"/>
<!-- ServiceCollection + BuildServiceProvider for the CreateDataSource factory test. -->
<PackageReference Include="Microsoft.Extensions.DependencyInjection"/>
<!-- In-process gRPC server capturing the published client's ReadEvents request (#675 bounds pin). -->
<PackageReference Include="Grpc.AspNetCore"/>
<PackageReference Include="xunit.runner.visualstudio">
<PrivateAssets>all</PrivateAssets>
<IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
</PackageReference>
</ItemGroup>
<ItemGroup>
<FrameworkReference Include="Microsoft.AspNetCore.App"/>
</ItemGroup>
<ItemGroup>
<ProjectReference Include="..\..\..\src\Drivers\ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway\ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.csproj"/>
</ItemGroup>
@@ -60,3 +60,10 @@ public sealed class QuestDbModifiedValueConformance(QuestDbFixture fx) : Modifie
/// <inheritdoc/>
protected override Task<IConformanceBackend> CreateBackendAsync() => QuestDbConformanceBackend.CreateAsync(fx);
}
[Collection(QuestDbCollection.Name)]
public sealed class QuestDbEventReadConformance(QuestDbFixture fx) : EventReadConformance
{
/// <inheritdoc/>
protected override Task<IConformanceBackend> CreateBackendAsync() => QuestDbConformanceBackend.CreateAsync(fx);
}
@@ -143,6 +143,27 @@ internal sealed class QuestDbConformanceBackend : IConformanceBackend
await Wal.WaitAsync(Conn.SampleTable, CancellationToken.None);
}
/// <inheritdoc/>
public async Task SeedEventsAsync(params (DateTime T, string Id, string Source)[] events)
{
await using (var c = await Conn.OpenAsync(CancellationToken.None))
{
foreach (var (t, id, source) in events)
{
await using var cmd = Conn.CreateCommand(c,
$"INSERT INTO {Conn.EventTable} (ts, event_id, source, severity, message, received_utc) VALUES ($1, $2, $3, $4, $5, $6)");
QuestDbConnection.BindTimestamp(cmd, t);
QuestDbConnection.BindText(cmd, id);
QuestDbConnection.BindText(cmd, source);
cmd.Parameters.Add(new NpgsqlParameter { NpgsqlDbType = NpgsqlDbType.Integer, Value = 500 });
QuestDbConnection.BindText(cmd, null);
QuestDbConnection.BindTimestamp(cmd, t);
await cmd.ExecuteNonQueryAsync();
}
}
await Wal.WaitAsync(Conn.EventTable, CancellationToken.None);
}
/// <summary>Runs a statement against this table set (test-only escape hatch for hand-made rows), then waits for WAL visibility of the sample table.</summary>
/// <param name="sql">The statement; parameters are bound positionally by <paramref name="bind"/>.</param>
/// <param name="bind">Binds the parameters.</param>
@@ -60,3 +60,10 @@ public sealed class SqlServerModifiedValueConformance(SqlHistorianFixture fx) :
/// <inheritdoc/>
protected override Task<IConformanceBackend> CreateBackendAsync() => SqlServerConformanceBackend.CreateAsync(fx);
}
[Collection(SqlHistorianCollection.Name)]
public sealed class SqlServerEventReadConformance(SqlHistorianFixture fx) : EventReadConformance
{
/// <inheritdoc/>
protected override Task<IConformanceBackend> CreateBackendAsync() => SqlServerConformanceBackend.CreateAsync(fx);
}
@@ -40,6 +40,22 @@ internal sealed class SqlServerConformanceBackend : IConformanceBackend
public async Task SeedAsync(string tag, params (DateTime T, HistorianValue V, uint Status)[] samples) =>
await _db.SeedAsync(tag, samples);
/// <inheritdoc/>
public async Task SeedEventsAsync(params (DateTime T, string Id, string Source)[] events)
{
await using var c = await _db.Conn.OpenAsync(CancellationToken.None);
foreach (var (t, id, source) in events)
{
await using var cmd = _db.Conn.CreateCommand(c,
$"INSERT {_db.Conn.Q("Event")} (EventId, SourceName, AlarmId, AlarmName, AlarmTypeName, Severity, EventKind, Message, " +
"UserName, Comment, EventTimeUtc, ReceivedTimeUtc) VALUES (@id, @src, 'a', 'n', 'T', 500, 'Activated', NULL, NULL, NULL, @t, @t)");
cmd.Parameters.AddWithValue("@id", id);
cmd.Parameters.AddWithValue("@src", source);
SqlHistorianConnection.AddDateTime2(cmd, "@t", t);
await cmd.ExecuteNonQueryAsync();
}
}
/// <inheritdoc/>
public async Task<IReadOnlyList<(DateTime T, HistorianValue V, uint Status)>> ReadBackAsync(string tag)
{
@@ -54,8 +54,48 @@ public sealed class EventHistoryPagingOverTheWireTests
finally { await rig.DisposeAsync(); }
}
[Fact]
public async Task A_capped_REVERSE_event_read_pages_newest_first_through_the_session_store_exactly_once()
{
// Gitea #675: StartTime later than EndTime. Before the fix the cap kept the EARLIEST 10, ascending, with no
// continuation point.
var ct = TestContext.Current.CancellationToken;
var rig = await StartAsync("EvtPagingWireRev", encrypted: false, ct);
try
{
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>();
var sizes = new List<int>();
byte[]? cp = null;
do
{
var (ids, next) = await ReadPageAsync(session, node, cp, ct, reverse: true);
collected.AddRange(ids);
sizes.Add(ids.Count);
cp = next;
sizes.Count.ShouldBeLessThan(30, "paging must terminate");
}
while (cp is not null);
sizes.ShouldBe([10, 10, 5]);
// Newest-first: time descending, ties by EventId descending.
collected.ShouldBe(events.OrderByDescending(e => e.EventTimeUtc)
.ThenByDescending(e => e.EventId, StringComparer.Ordinal).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)
ISession session, NodeId node, byte[]? cp, CancellationToken ct, bool reverse = false)
{
var filter = new EventFilter();
filter.SelectClauses.Add(new SimpleAttributeOperand(ObjectTypeIds.BaseEventType, new QualifiedName("EventId"))
@@ -64,8 +104,8 @@ public sealed class EventHistoryPagingOverTheWireTests
});
var details = new ReadEventDetails
{
StartTime = Epoch.AddDays(-1),
EndTime = Epoch.AddDays(1),
StartTime = reverse ? Epoch.AddDays(1) : Epoch.AddDays(-1),
EndTime = reverse ? Epoch.AddDays(-1) : Epoch.AddDays(1),
NumValuesPerNode = 10,
Filter = filter,
};
@@ -94,6 +134,18 @@ public sealed class EventHistoryPagingOverTheWireTests
return new HistoricalEventsResult(window.Take(cap).ToList(), window.Count > cap ? new byte[] { 1 } : null);
}
/// <summary>The #675 overload: reverse is <c>(lo, hi]</c> by (time, id) descending.</summary>
public async Task<HistoricalEventsResult> ReadEventsAsync(
string? sourceName, DateTime startUtc, DateTime endUtc, int maxEvents, bool reverse, CancellationToken ct)
{
if (!reverse) return await ReadEventsAsync(sourceName, startUtc, endUtc, maxEvents, ct).ConfigureAwait(false);
await Task.Delay(1, ct).ConfigureAwait(false);
var (lo, hi) = startUtc <= endUtc ? (startUtc, endUtc) : (endUtc, startUtc);
var window = _all.Where(e => e.EventTimeUtc > lo && e.EventTimeUtc <= hi).Reverse().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(
@@ -199,11 +199,168 @@ public sealed class NodeManagerHistoryReadEventsPagingTests : IDisposable
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));
// A resumed page is FULL (#675): it over-fetches by the boundary skip it then trims.
mixed[0].Ids.ShouldBe(src.All.Skip(4).Take(4).Select(e => e.EventId));
mixed[1].Error.StatusCode.Code.ShouldBe(StatusCodes.BadContinuationPointInvalid);
await host.DisposeAsync();
}
// --- Gitea #675: reverse (newest-first) event paging ------------------------------------------------------
[Fact]
public async Task Reverse_events_page_newest_first_exactly_once_with_exact_page_counts()
{
var (host, nm, node, src) = await SetupAsync(Distinct(250));
var pages = new List<int>();
var ids = DrainAsync(nm, node, max: 100, start: Late, end: Early, pageSizes: pages);
ids.ShouldBe(src.All.Reverse().Select(e => e.EventId));
pages.ShouldBe([100, 100, 50]);
await host.DisposeAsync();
}
[Fact]
public async Task Reverse_first_page_is_the_LATEST_events_with_a_continuation_point()
{
// The pre-#675 shape: the cap kept the EARLIEST N, ascending, Good, and no continuation point.
var (host, nm, node, _) = await SetupAsync(Distinct(10));
var p1 = ReadEvents(nm, node, max: 3, inboundCp: null, end: Early, start: Late);
p1.Error.StatusCode.Code.ShouldBe(StatusCodes.Good);
p1.Ids.ShouldBe(["e0009", "e0008", "e0007"]);
p1.Cp.ShouldNotBeNull("a capped reverse read has more behind it and must page");
await host.DisposeAsync();
}
[Fact]
public async Task Reverse_tie_at_page_boundary_is_neither_duplicated_nor_skipped()
{
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);
// Newest-first, ties in descending EventId: cap 2 cuts the cluster at e4|e3.
DrainAsync(nm, node, max: 2, start: Late, end: Early).ShouldBe(["e5", "e4", "e3", "e2", "e1", "e0"]);
await host.DisposeAsync();
}
[Fact]
public async Task Reverse_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);
var pages = new List<int>();
DrainAsync(nm, node, max: 3, start: Late, end: Early, pageSizes: pages).ShouldBe(
["z0", "t6", "t5", "t4", "t3", "t2", "t1", "t0", "a0"]);
pages.Sum().ShouldBe(9);
await host.DisposeAsync();
}
[Fact]
public async Task Reverse_tie_cluster_pages_on_a_coarse_resolution_backend()
{
const long res = 10;
var events = new List<HistoricalEvent>();
DateTime At(int slot) => Epoch.AddTicks(slot * res);
events.Add(new("a0", "s", At(0), Epoch, "m", 500));
for (var i = 0; i < 7; i++) events.Add(new($"t{i}", "s", At(3), Epoch, "m", 500));
events.Add(new("z0", "s", At(8), Epoch, "m", 500));
var (host, nm, node, _) = await SetupAsync(events, res);
DrainAsync(nm, node, max: 3, start: Late, end: Early).ShouldBe(
["z0", "t6", "t5", "t4", "t3", "t2", "t1", "t0", "a0"]);
await host.DisposeAsync();
}
[Fact]
public async Task Reverse_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, end: Early, start: Late);
p1.Cp.ShouldNotBeNull();
ReadEvents(nm, node, max: 3, inboundCp: p1.Cp, end: Early, start: Late).Error.StatusCode.Code
.ShouldBe(StatusCodes.BadHistoryOperationUnsupported);
await host.DisposeAsync();
}
[Fact]
public async Task A_reverse_cursor_resumes_reverse_whatever_the_resume_request_says()
{
// The direction lives in the cursor: a resume carrying FORWARD details still continues newest-first.
var (host, nm, node, _) = await SetupAsync(Distinct(10));
var p1 = ReadEvents(nm, node, max: 4, inboundCp: null, end: Early, start: Late);
p1.Ids.ShouldBe(["e0009", "e0008", "e0007", "e0006"]);
var p2 = ReadEvents(nm, node, max: 4, inboundCp: p1.Cp); // default details: Early -> Late (forward)
p2.Error.StatusCode.Code.ShouldBe(StatusCodes.Good);
p2.Ids.ShouldBe(["e0005", "e0004", "e0003", "e0002"]);
await host.DisposeAsync();
}
[Fact]
public async Task A_forward_cursor_resumes_forward_whatever_the_resume_request_says()
{
var (host, nm, node, _) = await SetupAsync(Distinct(10));
var p1 = ReadEvents(nm, node, max: 4, inboundCp: null);
p1.Ids.ShouldBe(["e0000", "e0001", "e0002", "e0003"]);
var p2 = ReadEvents(nm, node, max: 4, inboundCp: p1.Cp, end: Early, start: Late);
p2.Ids.ShouldBe(["e0004", "e0005", "e0006", "e0007"]);
await host.DisposeAsync();
}
[Fact]
public async Task Reverse_unbounded_read_returns_everything_newest_first_without_paging()
{
var (host, nm, node, src) = await SetupAsync(Distinct(30));
var r = ReadEvents(nm, node, max: 0, inboundCp: null, end: Early, start: Late);
r.Ids.ShouldBe(src.All.Reverse().Select(e => e.EventId));
r.Cp.ShouldBeNull();
await host.DisposeAsync();
}
[Fact]
public async Task End_only_events_read_pages_backward_from_EndTime()
{
// StartTime unset + EndTime + NumValuesPerNode > 0 => backward from EndTime (Part 11; mirrors raw). The reverse
// window is (.., EndTime], so an event exactly AT EndTime is the first one returned.
var (host, nm, node, _) = await SetupAsync(Distinct(25));
var ids = DrainAsync(nm, node, max: 10, start: DateTime.MinValue, end: Epoch.AddSeconds(20));
ids.ShouldBe(Enumerable.Range(0, 21).Reverse().Select(i => $"e{i:0000}"));
await host.DisposeAsync();
}
[Fact]
public async Task A_backend_without_reverse_support_answers_BadHistoryOperationUnsupported_not_the_earliest_events()
{
var (host, nm, node, src) = await SetupAsync(Distinct(10));
nm.HistorianDataSource = new ForwardOnlyEventSource(src);
ReadEvents(nm, node, max: 3, inboundCp: null, end: Early, start: Late).Error.StatusCode.Code
.ShouldBe(StatusCodes.BadHistoryOperationUnsupported);
// Forward reads on the same backend are unaffected.
DrainAsync(nm, node, max: 4).ShouldBe(src.All.Select(e => e.EventId));
await host.DisposeAsync();
}
private static readonly DateTime Early = Epoch.AddDays(-1);
private static readonly DateTime Late = Epoch.AddDays(1);
/// <summary>A backend that implements only the five-argument (forward) event read - Gateway / Null shaped.</summary>
private sealed class ForwardOnlyEventSource(IHistorianDataSource inner) : IHistorianDataSource
{
public Task<HistoricalEventsResult> ReadEventsAsync(string? n, DateTime s, DateTime e, int m, CancellationToken ct) =>
inner.ReadEventsAsync(n, s, e, m, ct);
public Task<HistorianRead> ReadRawAsync(string r, DateTime s, DateTime e, uint m, CancellationToken ct) =>
inner.ReadRawAsync(r, s, e, m, ct);
public Task<HistorianRead> ReadProcessedAsync(string r, DateTime s, DateTime e, TimeSpan i, HistoryAggregateType a, CancellationToken ct) =>
inner.ReadProcessedAsync(r, s, e, i, a, ct);
public Task<HistorianRead> ReadAtTimeAsync(string r, IReadOnlyList<DateTime> t, CancellationToken ct) =>
inner.ReadAtTimeAsync(r, t, ct);
public HistorianHealthSnapshot GetHealthSnapshot() => inner.GetHealthSnapshot();
public void Dispose() { }
}
private sealed class RawSeriesSource : IHistorianDataSource
{
public Task<HistorianRead> ReadRawAsync(string r, DateTime s, DateTime e, uint m, CancellationToken ct)
@@ -265,16 +422,18 @@ public sealed class NodeManagerHistoryReadEventsPagingTests : IDisposable
return (started.Host, nm, nm.TryGetFolder(equipmentId)!.NodeId, src);
}
private static List<string> DrainAsync(OtOpcUaNodeManager nm, NodeId node, uint max, DateTime? end = null)
private static List<string> DrainAsync(
OtOpcUaNodeManager nm, NodeId node, uint max, DateTime? end = null, DateTime? start = null, List<int>? pageSizes = null)
{
var all = new List<string>();
byte[]? cp = null;
var pages = 0;
do
{
var r = ReadEvents(nm, node, max, cp, end);
var r = ReadEvents(nm, node, max, cp, end, start);
r.Error.StatusCode.Code.ShouldBe(StatusCodes.Good);
all.AddRange(r.Ids);
pageSizes?.Add(r.Ids.Count);
cp = r.Cp;
(++pages).ShouldBeLessThan(60, "paging must terminate");
}
@@ -283,11 +442,15 @@ public sealed class NodeManagerHistoryReadEventsPagingTests : IDisposable
}
private static (List<string> Ids, ServiceResult Error, byte[]? Cp) ReadEvents(
OtOpcUaNodeManager nm, NodeId node, uint max, byte[]? inboundCp, DateTime? end = null) =>
ReadBatch(nm, max, end, (node, inboundCp))[0];
OtOpcUaNodeManager nm, NodeId node, uint max, byte[]? inboundCp, DateTime? end = null, DateTime? start = null) =>
ReadBatch(nm, max, end, start, (node, inboundCp))[0];
private static List<(List<string> Ids, ServiceResult Error, byte[]? Cp)> ReadBatch(
OtOpcUaNodeManager nm, uint max, DateTime? end, params (NodeId Node, byte[]? Cp)[] reads)
OtOpcUaNodeManager nm, uint max, DateTime? end, params (NodeId Node, byte[]? Cp)[] reads) =>
ReadBatch(nm, max, end, null, reads);
private static List<(List<string> Ids, ServiceResult Error, byte[]? Cp)> ReadBatch(
OtOpcUaNodeManager nm, uint max, DateTime? end, DateTime? start, params (NodeId Node, byte[]? Cp)[] reads)
{
var context = new OperationContext(
new RequestHeader(), secureChannelContext: null, RequestType.HistoryRead, identity: null);
@@ -298,7 +461,7 @@ public sealed class NodeManagerHistoryReadEventsPagingTests : IDisposable
});
var details = new ReadEventDetails
{
StartTime = Epoch.AddDays(-1),
StartTime = start ?? Epoch.AddDays(-1),
EndTime = end ?? Epoch.AddDays(1),
NumValuesPerNode = max,
Filter = filter,
@@ -361,6 +524,22 @@ public sealed class NodeManagerHistoryReadEventsPagingTests : IDisposable
window.Take(cap).ToList(), truncated ? new byte[] { 1 } : null));
}
/// <summary>Like the SQL backends' #675 overload: reverse => <c>(lo, hi]</c> newest-first by (time, id) descending.</summary>
public Task<HistoricalEventsResult> ReadEventsAsync(
string? sourceName, DateTime startUtc, DateTime endUtc, int maxEvents, bool reverse, CancellationToken ct)
{
if (!reverse) return ReadEventsAsync(sourceName, startUtc, endUtc, maxEvents, ct);
Calls++;
var (lo, hi) = startUtc <= endUtc ? (startUtc, endUtc) : (endUtc, startUtc);
lo = Trunc(lo);
hi = Trunc(hi);
var window = All.Where(e => e.EventTimeUtc > lo && e.EventTimeUtc <= hi).Reverse().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(