diff --git a/docs/Historian.md b/docs/Historian.md index e95f1f10..971fb43d 100644 --- a/docs/Historian.md +++ b/docs/Historian.md @@ -632,13 +632,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) @@ -861,8 +879,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)). @@ -1205,7 +1225,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. diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/Historian/HistoryReadContracts.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/Historian/HistoryReadContracts.cs index 0be71792..cb98f97a 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/Historian/HistoryReadContracts.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/Historian/HistoryReadContracts.cs @@ -88,7 +88,7 @@ public sealed class HistoryReadLimitExceededException : InvalidOperationExceptio } /// -/// One row returned by — a historical +/// One row returned by — a historical /// alarm/event record, not the OPC UA live-event stream. Fields match the minimum set the /// Server needs to populate a HistoryEventFieldList for HistoryReadEvents responses. /// @@ -106,8 +106,8 @@ public sealed record HistoricalEvent( string? Message, ushort Severity); -/// Result of a call. -/// Events in chronological order by EventTimeUtc. +/// Result of a call. +/// Events ordered by EventTimeUtc in the read's direction — ascending, or descending for a reverse read (#675). /// Opaque token for the next call when more events are available; null when complete. public sealed record HistoricalEventsResult( IReadOnlyList Events, diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/Historian/IHistorianDataSource.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/Historian/IHistorianDataSource.cs index a135d765..a891fd70 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/Historian/IHistorianDataSource.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/Historian/IHistorianDataSource.cs @@ -79,7 +79,7 @@ public interface IHistorianDataSource : IDisposable /// uint maxValuesPerNode because their OPC UA HistoryRead surface has no /// equivalent "use default" sentinel. /// - /// This surface declares and + /// This surface declares and /// as required members — a server-side historian owns the full read surface. /// /// The source name to filter events, or null to return events from all sources. @@ -101,6 +101,46 @@ public interface IHistorianDataSource : IDisposable int maxEvents, CancellationToken cancellationToken); + /// + /// Read historical alarm / event records in a chosen direction — the overload the server's HistoryReadEvents + /// paging calls (Gitea #675). + /// + /// + /// + /// The bounds may be given in either order; let lo/hi be the earlier/later of them. Forward + /// ( false) is exactly : + /// [lo, hi) ascending by (EventTimeUtc, EventId). Reverse is the Part 11 reverse window + /// (lo, hi] in DESCENDING (EventTimeUtc, EventId) order, so a cap keeps the LATEST + /// 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. + /// and the continuation-point contract are as for the forward read. + /// + /// + /// The default forwards a forward read to the five-argument overload and throws + /// for a reverse one — a backend that cannot read newest-first (Gateway, + /// Null) — which the server reports as BadHistoryOperationUnsupported rather than returning the + /// earliest events in ascending order as if they were the answer. + /// + /// + /// The source name to filter events, or null to return events from all sources. + /// One bound of the window (UTC). + /// The other bound of the window (UTC). + /// The cap; non-positive uses the backend default (see the five-argument overload). + /// true to read newest-first over (lo, hi]. + /// Cancellation. + /// The matching events in the read's direction, with a continuation point if truncated. + /// A reverse read on a backend that cannot read newest-first. + Task 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); + /// /// Point-in-time health snapshot for diagnostics and dashboards. Pure /// observation; never blocks on backend I/O. diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/GatewayHistorianDataSource.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/GatewayHistorianDataSource.cs index 556bbde2..a3e2ebb5 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/GatewayHistorianDataSource.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/GatewayHistorianDataSource.cs @@ -125,6 +125,21 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis } } + /// + /// + /// A reverse (newest-first) read is refused with — the server reports + /// BadHistoryOperationUnsupported (Gitea #675). The gateway's ReadEvents streams ascending only and + /// rejects end < start, 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. + /// + public Task 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); + /// public async Task ReadEventsAsync( string? sourceName, DateTime startUtc, DateTime endUtc, int maxEvents, diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianDataSource.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianDataSource.cs index 3f6b939b..a79c976d 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianDataSource.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianDataSource.cs @@ -361,8 +361,18 @@ public sealed class QuestDbHistorianDataSource : IHistorianDataSource /// maxEvents + 1 rows are fetched; when the extra row arrives it is dropped and the continuation point is /// the UTF-8 of for that first not-returned event. /// + public Task ReadEventsAsync( + string? sourceName, DateTime startUtc, DateTime endUtc, int maxEvents, CancellationToken cancellationToken) => + ReadEventsAsync(sourceName, startUtc, endUtc, maxEvents, reverse: false, cancellationToken); + + /// + /// + /// Forward is the five-argument read. Reverse (Gitea #675) is the (rounded) (lo, hi] ordered by + /// (ts, event_id) DESCENDING, so LIMIT keeps the latest events; truncation and the continuation point + /// work as forward. + /// public async Task 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); diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerHistorianDataSource.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerHistorianDataSource.cs index e2c31a32..315009e4 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerHistorianDataSource.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerHistorianDataSource.cs @@ -320,8 +320,18 @@ public sealed class SqlServerHistorianDataSource : IHistorianDataSource /// of "{EventTimeUtc.Ticks}|{EventId}" of that first not-returned event (opaque to callers; only /// its presence is contractual). /// + public Task ReadEventsAsync( + string? sourceName, DateTime startUtc, DateTime endUtc, int maxEvents, CancellationToken cancellationToken) => + ReadEventsAsync(sourceName, startUtc, endUtc, maxEvents, reverse: false, cancellationToken); + + /// + /// + /// Forward is the five-argument read. Reverse (Gitea #675) is (lo, hi] ordered by + /// (EventTimeUtc, EventId) DESCENDING — the clustered index scanned backwards — so TOP keeps the latest + /// events; truncation and the continuation point work as forward. + /// public async Task 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. @@ -334,9 +344,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); diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/HistoryPaging.cs b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/HistoryPaging.cs index 8526d254..667d16a1 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/HistoryPaging.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/HistoryPaging.cs @@ -46,6 +46,11 @@ namespace ZB.MOM.WW.OtOpcUa.OpcUaServer; /// 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). /// +/// +/// only (Gitea #675): the read is newest-first over +/// (EndUtc, NextStartUtc]. A resumed events read takes its direction from here, never from the resume +/// request's details. Raw keeps deriving it from the window (NextStartUtc > EndUtc). +/// 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); /// The HistoryRead kind that issued a . internal enum HistoryCursorKind diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs index d7b1db61..3fdc8cfe 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs @@ -102,7 +102,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 /// / . private readonly Dictionary _alarmNotifierWiring = new(); /// Phase C: event-notifier folder NodeId-identifier → the event-history source - /// name passed to . The equipment-folder NodeId + /// name passed to . 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 /// only when a real historian is wired at promotion time, @@ -5195,6 +5195,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 var endUtc = details.EndTime; var numValuesPerNode = details.NumValuesPerNode; var boundarySkip = 0; + bool reverse; if (resumed) { @@ -5214,36 +5215,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 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; @@ -5256,10 +5280,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 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( @@ -5274,7 +5302,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(sliceCount); for (var i = sliceStart; i < sliceStart + sliceCount; i++) slice.Add(cluster[i]); @@ -5283,17 +5311,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); @@ -5314,6 +5345,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 @@ -5693,7 +5733,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 private static int ClampToInt(uint value) => value > int.MaxValue ? int.MaxValue : (int)value; /// Map a HistoryRead-Events NumValuesPerNode request cap onto the - /// maxEvents argument, honouring the OPC UA + /// maxEvents argument, honouring the OPC UA /// Part 4/11 semantics that NumValuesPerNode == 0 means "no limit — return ALL values". /// We translate 0 to UNBOUNDED () — a very large positive cap — rather than /// the backend's maxEvents <= 0 "use the default cap" sentinel, so a client asking for the diff --git a/tests/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions.Tests/Historian/IHistorianDataSourceContractTests.cs b/tests/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions.Tests/Historian/IHistorianDataSourceContractTests.cs index 458e6796..847c6bad 100644 --- a/tests/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions.Tests/Historian/IHistorianDataSourceContractTests.cs +++ b/tests/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions.Tests/Historian/IHistorianDataSourceContractTests.cs @@ -45,9 +45,10 @@ public sealed class IHistorianDataSourceContractTests [InlineData("ReadEventsAsync", typeof(Task))] 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); } /// Verifies that GetHealthSnapshot is synchronous. @@ -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); + } } /// @@ -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 Overloads(string name) => + typeof(IHistorianDataSource).GetMethods().Where(m => m.Name == name).ToList(); } diff --git a/tests/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions.Tests/Historian/NullHistorianDataSourceTests.cs b/tests/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions.Tests/Historian/NullHistorianDataSourceTests.cs index 6ed8afb7..e9045e7b 100644 --- a/tests/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions.Tests/Historian/NullHistorianDataSourceTests.cs +++ b/tests/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions.Tests/Historian/NullHistorianDataSourceTests.cs @@ -67,6 +67,20 @@ public sealed class NullHistorianDataSourceTests result.ContinuationPoint.ShouldBeNull(); } + /// + /// #675: the direction-aware overload's default — forward is the five-argument read, reverse is refused + /// (the server answers BadHistoryOperationUnsupported). + /// + [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(() => + source.ReadEventsAsync(null, DateTime.UtcNow, DateTime.UtcNow.AddHours(-1), 0, reverse: true, Ct)); + } + /// GetHealthSnapshot reports a fully-disabled (zeroed, disconnected, no-node) shape. [Fact] public void GetHealthSnapshot_is_disabled() diff --git a/tests/Drivers/Historian.Conformance/EventReadConformance.cs b/tests/Drivers/Historian.Conformance/EventReadConformance.cs new file mode 100644 index 00000000..8d8110ca --- /dev/null +++ b/tests/Drivers/Historian.Conformance/EventReadConformance.cs @@ -0,0 +1,106 @@ +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Conformance; + +/// +/// The direction-aware event read (Gitea #675): forward is [lo, hi) ascending by (EventTimeUtc, EventId); +/// reverse is the Part 11 reverse window (lo, hi] 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. +/// +public abstract class EventReadConformance : ConformanceBase +{ + private static readonly CancellationToken Ct = CancellationToken.None; + + /// 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. + private async Task 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(); + 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"]); + } +} diff --git a/tests/Drivers/Historian.Conformance/IConformanceBackend.cs b/tests/Drivers/Historian.Conformance/IConformanceBackend.cs index 7e817533..afc9f69f 100644 --- a/tests/Drivers/Historian.Conformance/IConformanceBackend.cs +++ b/tests/Drivers/Historian.Conformance/IConformanceBackend.cs @@ -41,6 +41,13 @@ public interface IConformanceBackend : IAsyncDisposable /// The rows to store. Task SeedAsync(string tag, params (DateTime T, HistorianValue V, uint Status)[] samples); + /// + /// 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. + /// + /// The events to store. + Task SeedEventsAsync(params (DateTime T, string Id, string Source)[] events); + /// /// 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). diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayEventBoundsTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayEventBoundsTests.cs new file mode 100644 index 00000000..746e369d --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayEventBoundsTests.cs @@ -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; + +/// +/// Gitea #675: the Gateway backend's event-read bounds and direction, offline. +/// +/// A reverse (newest-first) read is refused before anything is sent — the gateway streams ascending only. +/// The open upper bound a start-only forward read uses (DateTime.MaxValue, 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. +/// +/// +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.Instance); + + await Should.ThrowAsync(() => + 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.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 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 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(); + await app.StartAsync(); + var port = new Uri(app.Urls.First()).Port; + return new Server(app, port); + } +} diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests.csproj b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests.csproj index 16e7fc33..a662c88e 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests.csproj +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests.csproj @@ -15,12 +15,18 @@ + + all runtime; build; native; contentfiles; analyzers; buildtransitive + + + + diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/QuestDbConformance.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/QuestDbConformance.cs index d0ca0bdd..a9987d1a 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/QuestDbConformance.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/QuestDbConformance.cs @@ -60,3 +60,10 @@ public sealed class QuestDbModifiedValueConformance(QuestDbFixture fx) : Modifie /// protected override Task CreateBackendAsync() => QuestDbConformanceBackend.CreateAsync(fx); } + +[Collection(QuestDbCollection.Name)] +public sealed class QuestDbEventReadConformance(QuestDbFixture fx) : EventReadConformance +{ + /// + protected override Task CreateBackendAsync() => QuestDbConformanceBackend.CreateAsync(fx); +} diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/QuestDbConformanceBackend.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/QuestDbConformanceBackend.cs index 1f3be1b3..ff4d6aad 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/QuestDbConformanceBackend.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/QuestDbConformanceBackend.cs @@ -143,6 +143,27 @@ internal sealed class QuestDbConformanceBackend : IConformanceBackend await Wal.WaitAsync(Conn.SampleTable, CancellationToken.None); } + /// + 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); + } + /// Runs a statement against this table set (test-only escape hatch for hand-made rows), then waits for WAL visibility of the sample table. /// The statement; parameters are bound positionally by . /// Binds the parameters. diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests/SqlServerConformance.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests/SqlServerConformance.cs index 790beffe..73b9862f 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests/SqlServerConformance.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests/SqlServerConformance.cs @@ -60,3 +60,10 @@ public sealed class SqlServerModifiedValueConformance(SqlHistorianFixture fx) : /// protected override Task CreateBackendAsync() => SqlServerConformanceBackend.CreateAsync(fx); } + +[Collection(SqlHistorianCollection.Name)] +public sealed class SqlServerEventReadConformance(SqlHistorianFixture fx) : EventReadConformance +{ + /// + protected override Task CreateBackendAsync() => SqlServerConformanceBackend.CreateAsync(fx); +} diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests/SqlServerConformanceBackend.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests/SqlServerConformanceBackend.cs index fd17aeaa..69afa55b 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests/SqlServerConformanceBackend.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests/SqlServerConformanceBackend.cs @@ -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); + /// + 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(); + } + } + /// public async Task> ReadBackAsync(string tag) { diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/EventHistoryPagingOverTheWireTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/EventHistoryPagingOverTheWireTests.cs index 2ba2bd8f..c3499953 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/EventHistoryPagingOverTheWireTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/EventHistoryPagingOverTheWireTests.cs @@ -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(); + var sizes = new List(); + 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 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); } + /// The #675 overload: reverse is (lo, hi] by (time, id) descending. + public async Task 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 ReadRawAsync(string r, DateTime s, DateTime e, uint m, CancellationToken ct) => NullHistorianDataSource.Instance.ReadRawAsync(r, s, e, m, ct); public Task ReadProcessedAsync( diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadEventsPagingTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadEventsPagingTests.cs index 6cd34534..8736fd0c 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadEventsPagingTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadEventsPagingTests.cs @@ -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(); + 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 + { + 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 { 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(); + 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(); + 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(); + 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); + + /// A backend that implements only the five-argument (forward) event read - Gateway / Null shaped. + private sealed class ForwardOnlyEventSource(IHistorianDataSource inner) : IHistorianDataSource + { + public Task ReadEventsAsync(string? n, DateTime s, DateTime e, int m, CancellationToken ct) => + inner.ReadEventsAsync(n, s, e, m, ct); + public Task ReadRawAsync(string r, DateTime s, DateTime e, uint m, CancellationToken ct) => + inner.ReadRawAsync(r, s, e, m, ct); + public Task ReadProcessedAsync(string r, DateTime s, DateTime e, TimeSpan i, HistoryAggregateType a, CancellationToken ct) => + inner.ReadProcessedAsync(r, s, e, i, a, ct); + public Task ReadAtTimeAsync(string r, IReadOnlyList t, CancellationToken ct) => + inner.ReadAtTimeAsync(r, t, ct); + public HistorianHealthSnapshot GetHealthSnapshot() => inner.GetHealthSnapshot(); + public void Dispose() { } + } + private sealed class RawSeriesSource : IHistorianDataSource { public Task 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 DrainAsync(OtOpcUaNodeManager nm, NodeId node, uint max, DateTime? end = null) + private static List DrainAsync( + OtOpcUaNodeManager nm, NodeId node, uint max, DateTime? end = null, DateTime? start = null, List? pageSizes = null) { var all = new List(); 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 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 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 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)); } + /// Like the SQL backends' #675 overload: reverse => (lo, hi] newest-first by (time, id) descending. + public Task 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 ReadRawAsync(string r, DateTime s, DateTime e, uint m, CancellationToken ct) => NullHistorianDataSource.Instance.ReadRawAsync(r, s, e, m, ct); public Task ReadProcessedAsync(