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

fix(historian): #686 serve reverse/end-only raw reads on forward-only backends (Gateway)

Browse Source 
#669 turned an end-only capped HistoryRead-Raw into a reverse read (start > end)
for every backend; the gateway rejects start > end (InvalidArgument), so on the
default Gateway backend every end-only capped and every explicit reverse raw read
failed the node and was counted as a backend health failure.

- IHistorianDataSource.SupportsReverseRaw (default false; SqlServer/QuestDB/Null
  true, Gateway false; guard test pins that every production backend declares it).
- OtOpcUaNodeManager sends a reverse window only to a backend that declares it;
  otherwise ReverseRawReader walks (lo, hi] back in bounded forward segment reads
  (1 min, x4 while sparse; each <= MaxTieClusterOverfetch + 1 rows; a full read is
  re-read narrower, never answered truncated; filtered on sample timestamps, which
  also drops AVEVA's fabricated query-start rows; stops at the Unix epoch). A
  one-timestamp tie cluster over the bound => BadTooManyOperations.
- ServeRawPagedAsync maps HistoryReadLimitExceededException to BadTooManyOperations.
- GatewayHistorianDataSource.ReadRawAsync refuses start > end unsent, without
  recording a health failure.
This commit is contained in:
Joseph Doherty
2026-10-12 14:42:27 -04:00
parent 70fb1f1fda
commit 410c8d8976
11 changed files with 490 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions.Tests/Historian/HistorianDataSourceReverseEventsGuardTests.cs
+16
View File
@@ -64,6 +64,22 @@ public sealed class HistorianDataSourceReverseEventsGuardTests
}
}
[Fact]
public void Every_implementation_declares_its_reverse_raw_capability_itself()
{
// Gitea #686: the default (false) is safe — the server walks back with forward reads — but a backend that serves
// reverse natively and inherited it would silently lose its native path, and a wrapper of the Gateway that claimed
// true would send start > end to a gateway that rejects it. Declare it, both ways.
foreach (var type in ProductionImplementations())
{
var map = type.GetInterfaceMap(typeof(IHistorianDataSource));
var getter = map.TargetMethods[Array.FindIndex(map.InterfaceMethods,
m => m.Name == "get_" + nameof(IHistorianDataSource.SupportsReverseRaw))];
getter.DeclaringType.ShouldBe(type,
$"{type.FullName} must declare SupportsReverseRaw itself, not inherit the interface default");
}
}
[Fact]
public void Every_implementation_declares_its_supported_aggregates_itself()
{
tests/Drivers/Historian.Conformance/RawReadConformance.cs
+8
View File
@@ -56,6 +56,14 @@ public abstract class RawReadConformance : ConformanceBase
Values(await B.CreateDataSource().ReadRawAsync("A", At(10), max, 0, Ct)).ShouldBe([20d, 30d, 40d, 50d]);
}
[Fact]
public void The_backend_reports_reverse_raw_support()
{
// #686: the server only sends start > end to a backend that declares it; these backends serve it natively.
using var src = B.CreateDataSource();
src.SupportsReverseRaw.ShouldBeTrue();
}
[Fact]
public async Task Reverse_read_is_descending_over_end_exclusive_start_inclusive()
{
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayRawDirectionTests.cs
+45
View File
@@ -0,0 +1,45 @@
using Microsoft.Extensions.Logging.Abstractions;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests;
/// <summary>
/// Gitea #686: the gateway's <c>ReadRaw</c> streams ascending only and rejects <c>start &gt; end</c> with
/// <c>InvalidArgument</c>, which used to be recorded as a backend health failure. The source declares it cannot read
/// newest-first (so the server walks back with forward reads), and refuses a reverse window itself without sending
/// anything or counting a failure.
/// </summary>
public sealed class GatewayRawDirectionTests
{
private static CancellationToken Ct => TestContext.Current.CancellationToken;
[Fact]
public void The_gateway_source_declares_it_cannot_read_raw_newest_first()
{
IHistorianDataSource ds = new GatewayHistorianDataSource(
new FakeHistorianGatewayClient(), NullLogger<GatewayHistorianDataSource>.Instance);
ds.SupportsReverseRaw.ShouldBeFalse();
}
[Fact]
public async Task A_reverse_raw_window_is_refused_unsent_and_is_not_a_backend_failure()
{
var fake = new FakeHistorianGatewayClient();
IHistorianDataSource ds = new GatewayHistorianDataSource(fake, NullLogger<GatewayHistorianDataSource>.Instance);
var end = new DateTime(2026, 10, 12, 18, 7, 45, DateTimeKind.Utc);
await Should.ThrowAsync<NotSupportedException>(() =>
ds.ReadRawAsync("Gate681", end, new DateTime(0, DateTimeKind.Utc), 3, Ct));
fake.ReadRawCallCount.ShouldBe(0);
var health = ds.GetHealthSnapshot();
health.TotalFailures.ShouldBe(0);
health.ConsecutiveFailures.ShouldBe(0);
// Positive control: a forward read of the same window is sent.
await ds.ReadRawAsync("Gate681", new DateTime(2026, 10, 12, 18, 0, 0, DateTimeKind.Utc), end, 3, Ct);
fake.ReadRawCallCount.ShouldBe(1);
}
}
tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadPagingTests.cs
+232
View File
@@ -466,6 +466,169 @@ public sealed class NodeManagerHistoryReadPagingTests : IDisposable
await host.DisposeAsync();
}
// --- #686: reverse / end-only raw reads on a backend that only reads forward (the Gateway) -------------------------
/// <summary>Boots a node manager over a <see cref="ForwardOnlyHistorianDataSource"/> and one historized variable.</summary>
private async Task<(OpcUaApplicationHost Host, OtOpcUaNodeManager Nm, NodeId NodeId)> BootForwardOnlyAsync(
ForwardOnlyHistorianDataSource source, string name)
{
var (host, server) = await BootAsync();
var nm = server.NodeManager!;
nm.HistoryContinuationStore = new InMemoryHistoryContinuationStore();
nm.HistorianDataSource = source;
nm.EnsureVariable($"eq-1/{name}", parentFolderNodeId: null, displayName: name, dataType: "Double",
writable: false, historianTagname: $"WW.{name}", realm: AddressSpaceRealm.Uns);
return (host, nm, nm.TryGetVariable($"eq-1/{name}")!.NodeId);
}
/// <summary>
/// #686: an end-only capped read (Part 11: page BACKWARD from EndTime) on a backend that cannot read newest-first
/// pages every sample newest-first — and never sends that backend <c>start &gt; end</c> (the gateway rejects it
/// <c>InvalidArgument</c>, which used to fail the node and count as a backend fault).
/// </summary>
[Fact]
public async Task Raw_end_only_read_on_a_forward_only_backend_pages_descending_without_a_reverse_window()
{
var source = new ForwardOnlyHistorianDataSource(MakeSeries(count: 250, stepSeconds: 1));
var (host, nm, nodeId) = await BootForwardOnlyAsync(source, "FwdEndOnly");
var collected = await DrainAsync(nm, nodeId, DateTime.MinValue, Epoch.AddHours(1), max: 100);
collected.ShouldBe(Enumerable.Range(0, 250).Reverse().Select(i => (double)i));
source.ReverseWindowCount.ShouldBe(0);
await host.DisposeAsync();
}
/// <summary>#686: the first page of an end-only capped read is the N values AT OR BEFORE EndTime, newest first —
/// not the oldest N in history (the pre-#669 forward meaning), and not the synthetic boundary row the AVEVA-shaped
/// fake emits at each query start.</summary>
[Fact]
public async Task Raw_end_only_capped_read_on_a_forward_only_backend_returns_the_latest_values_before_end_time()
{
var source = new ForwardOnlyHistorianDataSource(MakeSeries(count: 250, stepSeconds: 1));
var (host, nm, nodeId) = await BootForwardOnlyAsync(source, "FwdLatest");
var (values, error, cp) = ReadRaw(nm, nodeId, DateTime.MinValue, Epoch.AddSeconds(100), max: 3, inboundCp: null);
error.StatusCode.Code.ShouldBe(StatusCodes.Good);
values.ShouldBe([100d, 99d, 98d]);
cp.ShouldNotBeNull();
source.ReverseWindowCount.ShouldBe(0);
await host.DisposeAsync();
}
/// <summary>#686: an explicit reverse read with no cap returns the whole window newest-first in one page.</summary>
[Fact]
public async Task Raw_unbounded_explicit_reverse_read_on_a_forward_only_backend_returns_the_window_descending()
{
var source = new ForwardOnlyHistorianDataSource(MakeSeries(count: 250, stepSeconds: 1));
var (host, nm, nodeId) = await BootForwardOnlyAsync(source, "FwdRevAll");
// (end, start] = (Epoch+9s, Epoch+200s] — both edges exercised.
var (values, error, cp) = ReadRaw(nm, nodeId, Epoch.AddSeconds(200), Epoch.AddSeconds(9), max: 0, inboundCp: null);
error.StatusCode.Code.ShouldBe(StatusCodes.Good);
values.ShouldBe(Enumerable.Range(10, 191).Reverse().Select(i => (double)i));
cp.ShouldBeNull();
source.ReverseWindowCount.ShouldBe(0);
await host.DisposeAsync();
}
/// <summary>#686: a reverse read paging through a tie cluster on a forward-only backend — every value once, newest first.</summary>
[Fact]
public async Task Raw_reverse_read_on_a_forward_only_backend_pages_through_a_tie_cluster()
{
// No AVEVA boundary row here: paging WITHIN the cluster re-reads it with the node manager's start == end point
// over-fetch, and a row fabricated at a point read's start is indistinguishable from a tie at that instant —
// dropping it is the Gateway source's job (#687), not the walk's.
var source = new ForwardOnlyHistorianDataSource(
MakeSeriesWithTieCluster(distinctBefore: 3, tieCount: 5, distinctAfter: 2), emitBoundaryRow: false);
var (host, nm, nodeId) = await BootForwardOnlyAsync(source, "FwdRevBurst");
var collected = await DrainAsync(nm, nodeId, start: Epoch.AddHours(1), end: Epoch.AddSeconds(-1), max: 2);
collected.ShouldBe(Enumerable.Range(0, 10).Reverse().Select(i => (double)i));
source.ReverseWindowCount.ShouldBe(0);
await host.DisposeAsync();
}
/// <summary>#686: a segment denser than the per-read row budget (<c>MaxTieClusterOverfetch</c>) is narrowed and re-read,
/// never answered from a truncated forward read.</summary>
[Fact]
public async Task Raw_reverse_read_on_a_forward_only_backend_narrows_a_segment_over_the_row_budget()
{
var source = new ForwardOnlyHistorianDataSource(MakeSeries(count: 250, stepSeconds: 1));
var (host, nm, nodeId) = await BootForwardOnlyAsync(source, "FwdDense");
nm.MaxTieClusterOverfetch = 16; // a 1-minute segment holds 60 samples
var collected = await DrainAsync(nm, nodeId, DateTime.MinValue, Epoch.AddHours(1), max: 100);
collected.ShouldBe(Enumerable.Range(0, 250).Reverse().Select(i => (double)i));
source.MaxRowsReturned.ShouldBeLessThanOrEqualTo(17);
source.ReverseWindowCount.ShouldBe(0);
await host.DisposeAsync();
}
/// <summary>#686: sparse history far before EndTime is found, in a bounded number of forward reads.</summary>
[Fact]
public async Task Raw_end_only_read_on_a_forward_only_backend_finds_sparse_history_in_bounded_reads()
{
static DataValueSnapshot At(int year, double v)
{
var t = new DateTime(year, 1, 1, 0, 0, 0, DateTimeKind.Utc);
return new DataValueSnapshot(v, StatusCodes.Good, t, t);
}
var source = new ForwardOnlyHistorianDataSource([At(1990, 1), At(2000, 2)]);
var (host, nm, nodeId) = await BootForwardOnlyAsync(source, "FwdSparse");
var (values, error, cp) = ReadRaw(nm, nodeId, DateTime.MinValue, Epoch, max: 5, inboundCp: null);
error.StatusCode.Code.ShouldBe(StatusCodes.Good);
values.ShouldBe([2d, 1d]);
cp.ShouldBeNull();
source.ReadCount.ShouldBeLessThan(40);
source.ReverseWindowCount.ShouldBe(0);
await host.DisposeAsync();
}
/// <summary>#686: a single-timestamp tie cluster larger than the row budget cannot be read newest-first from
/// forward reads — the node is refused <c>BadTooManyOperations</c> (a request limit, not a backend fault).</summary>
[Fact]
public async Task Raw_reverse_read_on_a_forward_only_backend_refuses_a_tie_cluster_over_the_row_budget()
{
var source = new ForwardOnlyHistorianDataSource(MakeSeriesWithTieCluster(distinctBefore: 1, tieCount: 10, distinctAfter: 0));
var (host, nm, nodeId) = await BootForwardOnlyAsync(source, "FwdHugeTie");
nm.MaxTieClusterOverfetch = 4;
var (_, error, cp) = ReadRaw(nm, nodeId, Epoch.AddHours(1), Epoch.AddSeconds(-1), max: 2, inboundCp: null);
error.StatusCode.Code.ShouldBe(StatusCodes.BadTooManyOperations);
cp.ShouldBeNull();
source.ReverseWindowCount.ShouldBe(0);
await host.DisposeAsync();
}
/// <summary>A backend that DECLARES native reverse support is still sent the reverse window — one read.</summary>
[Fact]
public async Task Raw_end_only_read_on_a_reverse_capable_backend_is_one_native_reverse_read()
{
var (host, server) = await BootAsync();
var nm = server.NodeManager!;
nm.HistoryContinuationStore = new InMemoryHistoryContinuationStore();
var source = new SeriesHistorianDataSource(MakeSeries(count: 250, stepSeconds: 1));
nm.HistorianDataSource = source;
nm.EnsureVariable("eq-1/native", parentFolderNodeId: null, displayName: "Native", dataType: "Double",
writable: false, historianTagname: "WW.Native", realm: AddressSpaceRealm.Uns);
var nodeId = nm.TryGetVariable("eq-1/native")!.NodeId;
var (values, _, _) = ReadRaw(nm, nodeId, DateTime.MinValue, Epoch.AddSeconds(100), max: 3, inboundCp: null);
values.ShouldBe([100d, 99d, 98d]);
source.ReadCount.ShouldBe(1);
await host.DisposeAsync();
}
/// <summary>Pages a Raw read to completion and returns the union of the pages in arrival order.</summary>
private static Task<List<double>> DrainAsync(OtOpcUaNodeManager nm, NodeId nodeId, DateTime start, DateTime end, uint max)
{
@@ -584,6 +747,9 @@ public sealed class NodeManagerHistoryReadPagingTests : IDisposable
private sealed class SeriesHistorianDataSource(IReadOnlyList<DataValueSnapshot> series) : IHistorianDataSource
{
private int _readCount;
/// <summary>Serves start &gt; end natively — the SQL backends' contract (#686: declared, like they do).</summary>
public bool SupportsReverseRaw => true;
public int ReadCount => _readCount;
public void ResetReadCount() => _readCount = 0;
@@ -629,6 +795,70 @@ public sealed class NodeManagerHistoryReadPagingTests : IDisposable
public void Dispose() { }
}
/// <summary>
/// #686: a Gateway-shaped backend that reads FORWARD only. <see cref="SupportsReverseRaw"/> is false; a reverse window
/// (<c>start &gt; end</c>) is counted and rejected as the gateway does (<c>InvalidArgument</c>). A forward read is
/// inclusive <c>[start, end]</c> and, like an AVEVA <c>Full</c> read, first emits a synthetic boundary row stamped at
/// the query start carrying the value in effect there (Bad 0 when there is none) — counted against the cap.
/// </summary>
private sealed class ForwardOnlyHistorianDataSource(
IReadOnlyList<DataValueSnapshot> series, bool emitBoundaryRow = true) : IHistorianDataSource
{
private int _readCount;
private int _reverseWindowCount;
private int _maxRowsReturned;
public int ReadCount => _readCount;
public int ReverseWindowCount => _reverseWindowCount;
public int MaxRowsReturned => _maxRowsReturned;
public bool SupportsReverseRaw => false;
public Task<HistorianRead> ReadRawAsync(
string fullReference, DateTime startUtc, DateTime endUtc, uint maxValuesPerNode,
CancellationToken cancellationToken)
{
Interlocked.Increment(ref _readCount);
if (startUtc > endUtc)
{
Interlocked.Increment(ref _reverseWindowCount);
throw new InvalidOperationException("start_time must be less than or equal to end_time. (InvalidArgument)");
}
static DateTime Ts(DataValueSnapshot s) => s.SourceTimestampUtc ?? DateTime.MinValue;
var prior = series.LastOrDefault(s => Ts(s) <= startUtc);
var rows = new List<DataValueSnapshot>();
if (emitBoundaryRow)
rows.Add(prior is null
? new DataValueSnapshot(0d, StatusCodes.BadNoData, startUtc, startUtc)
: new DataValueSnapshot(prior.Value, prior.StatusCode, startUtc, startUtc));
rows.AddRange(series.Where(s => Ts(s) >= startUtc && Ts(s) <= endUtc));
var page = maxValuesPerNode == 0 ? rows : rows.Take((int)maxValuesPerNode).ToList();
int seen;
while (page.Count > (seen = Volatile.Read(ref _maxRowsReturned))
&& Interlocked.CompareExchange(ref _maxRowsReturned, page.Count, seen) != seen) { }
return Task.FromResult(new HistorianRead(page, null));
}
public Task<HistorianRead> ReadProcessedAsync(
string fullReference, DateTime startUtc, DateTime endUtc, TimeSpan interval,
HistoryAggregateType aggregate, CancellationToken cancellationToken) =>
Task.FromResult(new HistorianRead(Array.Empty<DataValueSnapshot>(), null));
public Task<HistorianRead> ReadAtTimeAsync(
string fullReference, IReadOnlyList<DateTime> timestampsUtc, CancellationToken cancellationToken) =>
Task.FromResult(new HistorianRead(Array.Empty<DataValueSnapshot>(), null));
public Task<HistoricalEventsResult> ReadEventsAsync(
string? sourceName, DateTime startUtc, DateTime endUtc, int maxEvents,
CancellationToken cancellationToken) =>
Task.FromResult(new HistoricalEventsResult(Array.Empty<HistoricalEvent>(), null));
public HistorianHealthSnapshot GetHealthSnapshot() => NullHistorianDataSource.Instance.GetHealthSnapshot();
public void Dispose() { }
}
/// <summary>
/// A microsecond-resolution backend, shaped like QuestDB: every inbound window bound is TRUNCATED to 10 ticks
/// (as <c>QuestDbConnection.Round</c> does) before the series is filtered, and the source reports
@@ -640,6 +870,8 @@ public sealed class NodeManagerHistoryReadPagingTests : IDisposable
public long TimestampResolutionTicks => 10;
public bool SupportsReverseRaw => true; // QuestDB serves reverse natively
private static DateTime Truncate(DateTime t) => new(t.Ticks - t.Ticks % 10, DateTimeKind.Utc);
public Task<HistorianRead> ReadRawAsync(