fix(historian): #675 review follow-ups — SupportsReverseEvents capability, no catch-as-unsupported

- IHistorianDataSource.SupportsReverseEvents (default false; SqlServer/QuestDB true; Gateway/Null declare false).
  The node manager checks it BEFORE calling the source: an end-only capped events read becomes reverse only on
  a supporting backend (Gateway, the default, keeps the pre-#675 forward [MinValue, EndTime) paging), and an
  explicit reverse read on a non-supporting backend is BadHistoryOperationUnsupported without a call.
- Removed the broad catch (NotSupportedException) in the events path: a NotSupportedException from inside a
  backend is now logged as a backend fault (Error), not an 'unsupported' Warning.
- Null declares the reverse overload itself; guard test: every production IHistorianDataSource declares both
  the reverse overload and the capability (positive-control scan of all four backends).
- Conformance: capability reported + oversized reverse tie cluster (forward over-fetch reversed == reverse
  read order; capped reverse read stopping inside the cluster flags truncation) on SqlServer + QuestDB.
- docs/Historian.md: end-only capped events read is backward on SqlServer/QuestDB, forward on Gateway.
This commit is contained in:
Joseph Doherty
2026-10-11 11:36:12 -04:00
parent 8e6ac4b344
commit 93d6c1da67
12 changed files with 290 additions and 27 deletions
@@ -0,0 +1,66 @@
using System.Reflection;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
namespace ZB.MOM.WW.OtOpcUa.Core.Abstractions.Tests.Historian;
/// <summary>
/// Gitea #675 review: the reverse-events surface is two interface members with defaults
/// (<see cref="IHistorianDataSource.SupportsReverseEvents"/> = false and the direction-aware
/// <c>ReadEventsAsync(…, bool reverse, …)</c>). A future wrapper / decorator that forwards only the five-argument read
/// and the other members would compile and silently fall back to the defaults — turning every reverse read into
/// <c>BadHistoryOperationUnsupported</c> on a backend that supports it. So every production implementation must declare
/// BOTH members itself.
/// </summary>
public sealed class HistorianDataSourceReverseEventsGuardTests
{
private static IReadOnlyList<Type> ProductionImplementations()
{
// Every ZB.MOM.WW.OtOpcUa production assembly next to the tests (this project references every historian backend).
var dir = AppContext.BaseDirectory;
var assemblies = Directory.GetFiles(dir, "ZB.MOM.WW.OtOpcUa.*.dll")
.Where(f => !Path.GetFileName(f).Contains(".Tests", StringComparison.Ordinal)
&& !Path.GetFileName(f).Contains("IntegrationTests", StringComparison.Ordinal))
.Select(f => Assembly.LoadFrom(f));
return assemblies
.SelectMany(a =>
{
try { return a.GetTypes(); }
catch (ReflectionTypeLoadException e) { return e.Types.Where(t => t is not null).Cast<Type>(); }
})
.Where(t => t is { IsClass: true, IsAbstract: false } && typeof(IHistorianDataSource).IsAssignableFrom(t))
.ToList();
}
[Fact]
public void The_scan_finds_every_known_backend()
{
// Positive control: a scan that finds nothing would pass the guard below vacuously.
var names = ProductionImplementations().Select(t => t.Name).ToList();
names.ShouldContain("NullHistorianDataSource");
names.ShouldContain("GatewayHistorianDataSource");
names.ShouldContain("SqlServerHistorianDataSource");
names.ShouldContain("QuestDbHistorianDataSource");
}
[Fact]
public void Every_implementation_declares_the_reverse_overload_and_the_capability_itself()
{
foreach (var type in ProductionImplementations())
{
var map = type.GetInterfaceMap(typeof(IHistorianDataSource));
MethodInfo Impl(Func<MethodInfo, bool> pick) =>
map.TargetMethods[Array.FindIndex(map.InterfaceMethods, m => pick(m))];
var reverseRead = Impl(m => m.Name == nameof(IHistorianDataSource.ReadEventsAsync)
&& m.GetParameters().Any(p => p.Name == "reverse"));
reverseRead.DeclaringType.ShouldBe(type,
$"{type.FullName} must declare ReadEventsAsync(…, bool reverse, …) itself, not inherit the interface default");
var capability = Impl(m => m.Name == "get_" + nameof(IHistorianDataSource.SupportsReverseEvents));
capability.DeclaringType.ShouldBe(type,
$"{type.FullName} must declare SupportsReverseEvents itself, not inherit the interface default");
}
}
}
@@ -96,6 +96,42 @@ public abstract class EventReadConformance : ConformanceBase
}
}
[Fact]
public void The_backend_reports_reverse_support()
{
using var src = B.CreateDataSource();
src.SupportsReverseEvents.ShouldBeTrue();
}
[Fact]
public async Task An_oversized_reverse_tie_cluster_sliced_the_way_the_server_does_matches_the_reverse_read_order()
{
// The server pages WITHIN a tie cluster larger than the page by over-fetching it FORWARD over [T, T + resolution)
// and reversing it for a reverse read (OtOpcUaNodeManager.ServeEventsAsync + HistoryPaging.SliceTieCluster). That is
// only correct if this backend's forward cluster, reversed, is exactly the order its reverse read emits the ties in
// — the tie-breaker must be the same key in both directions. Drive both through the real backend.
await B.SeedEventsAsync(Enumerable.Range(0, 12).Select(i => (At(2), $"c{i:00}", "s")).Append((At(3), "z", "s")).ToArray());
using var src = B.CreateDataSource();
var reverseOrder = (await src.ReadEventsAsync("s", At(10), At(0), 0, reverse: true, Ct)).Events
.Where(e => e.EventTimeUtc == B.Round(At(2))).Select(e => e.EventId).ToList();
var clusterReversed = (await src.ReadEventsAsync("s", B.Round(At(2)), B.Round(At(2)).AddTicks(B.ResolutionTicks), 1000, reverse: false, Ct))
.Events.Select(e => e.EventId).Reverse().ToList();
clusterReversed.Count.ShouldBe(12);
clusterReversed.ShouldBe(reverseOrder);
// And slicing it at a cap of 5 (skip 0, 5, 10) walks the cluster newest-first with no duplicate or gap.
var slices = new List<string>();
for (var skip = 0; skip < clusterReversed.Count; skip += 5) slices.AddRange(clusterReversed.Skip(skip).Take(5));
slices.ShouldBe(Enumerable.Range(0, 12).Reverse().Select(i => $"c{i:00}"));
// A capped reverse read stopping inside the cluster flags truncation, so the server's cluster branch is reachable.
var capped = await src.ReadEventsAsync("s", At(10), At(0), 4, reverse: true, Ct);
capped.Events.Select(e => e.EventId).ShouldBe(["z", "c11", "c10", "c09"]);
capped.ContinuationPoint.ShouldNotBeNull();
}
[Fact]
public async Task Reverse_read_with_a_null_source_spans_every_source()
{
@@ -35,6 +35,7 @@ public sealed class GatewayEventBoundsTests
{
var fake = new FakeHistorianGatewayClient();
IHistorianDataSource ds = new GatewayHistorianDataSource(fake, NullLogger<GatewayHistorianDataSource>.Instance);
ds.SupportsReverseEvents.ShouldBeFalse("the server must keep end-only reads forward on the default backend");
await Should.ThrowAsync<NotSupportedException>(() =>
ds.ReadEventsAsync("Pump1", DateTime.UtcNow, DateTime.UtcNow.AddHours(-1), 10, reverse: true, Ct));
@@ -134,6 +134,8 @@ public sealed class EventHistoryPagingOverTheWireTests
return new HistoricalEventsResult(window.Take(cap).ToList(), window.Count > cap ? new byte[] { 1 } : null);
}
public bool SupportsReverseEvents => true;
/// <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)
@@ -1,3 +1,4 @@
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Logging.Abstractions;
using Opc.Ua;
using Opc.Ua.Server;
@@ -346,6 +347,103 @@ public sealed class NodeManagerHistoryReadEventsPagingTests : IDisposable
private static readonly DateTime Early = Epoch.AddDays(-1);
private static readonly DateTime Late = Epoch.AddDays(1);
// --- #675 review follow-ups: the reverse-events capability ------------------------------------------------
[Fact]
public async Task End_only_capped_read_on_a_backend_without_reverse_support_pages_forward_as_before()
{
// Gateway (the default backend) cannot read newest-first: an end-only read keeps its pre-#675 forward
// [MinValue, EndTime) meaning there rather than turning into a refused reverse read.
var (host, nm, node, src) = await SetupAsync(Distinct(25));
var fwd = new NonReverseSource(src);
nm.HistorianDataSource = fwd;
var ids = DrainAsync(nm, node, max: 10, start: DateTime.MinValue, end: Epoch.AddSeconds(20));
ids.ShouldBe(Enumerable.Range(0, 20).Select(i => $"e{i:0000}"));
fwd.ReverseCalls.ShouldBe(0);
await host.DisposeAsync();
}
[Fact]
public async Task Explicit_reverse_read_on_a_backend_without_reverse_support_is_refused_without_calling_it()
{
var (host, nm, node, src) = await SetupAsync(Distinct(10));
var fwd = new NonReverseSource(src);
nm.HistorianDataSource = fwd;
ReadEvents(nm, node, max: 3, inboundCp: null, end: Early, start: Late).Error.StatusCode.Code
.ShouldBe(StatusCodes.BadHistoryOperationUnsupported);
fwd.ReverseCalls.ShouldBe(0, "the capability is checked BEFORE the source is called");
fwd.ForwardCalls.ShouldBe(0);
await host.DisposeAsync();
}
[Fact]
public async Task A_NotSupportedException_from_inside_a_supporting_backend_is_a_backend_fault_not_unsupported()
{
// A NotSupportedException thrown by a driver library (Npgsql, SqlClient) or the projection is a fault: it must be
// logged as an Error like any other backend failure, never downgraded to the "backend cannot read reverse" Warning.
var logger = new CapturingLogger();
var (host, nm, node, _) = await SetupAsync(Distinct(10), logger: logger);
nm.HistorianDataSource = new ThrowingReverseSource();
ReadEvents(nm, node, max: 3, inboundCp: null, end: Early, start: Late).Error.StatusCode.Code
.ShouldBe(StatusCodes.BadHistoryOperationUnsupported);
logger.Entries.ShouldContain(e => e.Level == LogLevel.Error && e.Exception is NotSupportedException);
logger.Entries.ShouldNotContain(e => e.Level == LogLevel.Warning);
await host.DisposeAsync();
}
/// <summary>Declares the reverse overload (so a call would be visible) but reports no reverse capability.</summary>
private sealed class NonReverseSource(IHistorianDataSource inner) : IHistorianDataSource
{
public int ReverseCalls { get; private set; }
public int ForwardCalls { get; private set; }
public bool SupportsReverseEvents => false;
public Task<HistoricalEventsResult> ReadEventsAsync(string? n, DateTime s, DateTime e, int m, CancellationToken ct)
{
ForwardCalls++;
return inner.ReadEventsAsync(n, s, e, m, ct);
}
public Task<HistoricalEventsResult> ReadEventsAsync(string? n, DateTime s, DateTime e, int m, bool reverse, CancellationToken ct)
{
if (reverse) { ReverseCalls++; return inner.ReadEventsAsync(n, s, e, m, true, ct); }
return 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() { }
}
/// <summary>Supports reverse, but its read fails inside with a library-level NotSupportedException.</summary>
private sealed class ThrowingReverseSource : IHistorianDataSource
{
public bool SupportsReverseEvents => true;
public Task<HistoricalEventsResult> ReadEventsAsync(string? n, DateTime s, DateTime e, int m, CancellationToken ct) =>
throw new NotSupportedException("driver: unsupported wire type");
public Task<HistoricalEventsResult> ReadEventsAsync(string? n, DateTime s, DateTime e, int m, bool reverse, CancellationToken ct) =>
throw new NotSupportedException("driver: unsupported wire type");
public Task<HistorianRead> ReadRawAsync(string r, DateTime s, DateTime e, uint m, CancellationToken ct) =>
NullHistorianDataSource.Instance.ReadRawAsync(r, s, e, m, ct);
public Task<HistorianRead> ReadProcessedAsync(string r, DateTime s, DateTime e, TimeSpan i, HistoryAggregateType a, CancellationToken ct) =>
NullHistorianDataSource.Instance.ReadProcessedAsync(r, s, e, i, a, ct);
public Task<HistorianRead> ReadAtTimeAsync(string r, IReadOnlyList<DateTime> t, CancellationToken ct) =>
NullHistorianDataSource.Instance.ReadAtTimeAsync(r, t, ct);
public HistorianHealthSnapshot GetHealthSnapshot() => NullHistorianDataSource.Instance.GetHealthSnapshot();
public void Dispose() { }
}
private sealed class CapturingLogger : ILogger<OtOpcUaNodeManager>
{
public System.Collections.Concurrent.ConcurrentQueue<(LogLevel Level, Exception? Exception, string Message)> Entries { get; } = new();
public IDisposable? BeginScope<TState>(TState state) where TState : notnull => null;
public bool IsEnabled(LogLevel logLevel) => true;
public void Log<TState>(LogLevel logLevel, EventId eventId, TState state, Exception? exception, Func<TState, Exception?, string> formatter) =>
Entries.Enqueue((logLevel, exception, formatter(state, exception)));
}
/// <summary>A backend that implements only the five-argument (forward) event read - Gateway / Null shaped.</summary>
private sealed class ForwardOnlyEventSource(IHistorianDataSource inner) : IHistorianDataSource
{
@@ -396,7 +494,7 @@ public sealed class NodeManagerHistoryReadEventsPagingTests : IDisposable
}
private async Task<(OpcUaApplicationHost Host, OtOpcUaNodeManager Nm, NodeId Node, SqlLikeEventSource Src)> SetupAsync(
List<HistoricalEvent> events, long resolution = 1)
List<HistoricalEvent> events, long resolution = 1, ILogger<OtOpcUaNodeManager>? logger = null)
{
var applicationUri = $"urn:OtOpcUa.EventPagingTest:{Guid.NewGuid():N}";
var started = await OpcUaHostStart.StartAsync(
@@ -410,7 +508,7 @@ public sealed class NodeManagerHistoryReadEventsPagingTests : IDisposable
PkiStoreRoot = _pkiRoot,
},
NullLogger<OpcUaApplicationHost>.Instance),
() => new OtOpcUaSdkServer(), Ct);
() => new OtOpcUaSdkServer(logger), Ct);
var nm = started.Server.NodeManager!;
nm.HistoryContinuationStore = new InMemoryHistoryContinuationStore();
var src = new SqlLikeEventSource(events, resolution);
@@ -503,6 +601,7 @@ public sealed class NodeManagerHistoryReadEventsPagingTests : IDisposable
private sealed class SqlLikeEventSource(List<HistoricalEvent> events, long resolution = 1) : IHistorianDataSource
{
public long TimestampResolutionTicks => resolution;
public bool SupportsReverseEvents => true;
private DateTime Trunc(DateTime d) => d.AddTicks(-(d.Ticks % resolution));
public IReadOnlyList<HistoricalEvent> All { get; } =