Joseph Doherty
9bb1363cd6
A tie cluster's truncated read left the width at one tick; once the gap above it was under two milliseconds nothing floored it, so segHi stepped down one tick per read: 30 rows at one ms 0.3 s below hi with a carried 1 s width took 10,006 identical reads (S8). (a) The top-anchored width is floored at the granularity, Min kept; (b) a segment no wider than the granularity is the unit that escalates to MaxTieClusterOverfetch and then refuses - per millisecond, exact for the ms-resolution Gateway (documented). Both are needed: the floor alone loops on crowded reads with a small tie budget. S8 is a repo test (4 shapes, reads <= 30, 0 repeated requests); random-data tests now measure the refusal unit as rows in any 1 ms window. Reviewer harness, 5,000 seeds x default/MSONLY/CROWD: worst page 70 reads (default/MSONLY), 63 (CROWD) at M=65536, p99.9 27-37; no page over 500 reads at any tie budget.
1279 lines
64 KiB
C#
1279 lines
64 KiB
C#
using Microsoft.Extensions.Logging.Abstractions;
|
||
using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
|
||
using Opc.Ua;
|
||
using Opc.Ua.Server;
|
||
using Shouldly;
|
||
using Xunit;
|
||
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
||
using HistorianRead = ZB.MOM.WW.OtOpcUa.Core.Abstractions.HistoryReadResult;
|
||
using SdkHistoryReadResult = Opc.Ua.HistoryReadResult;
|
||
|
||
namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests;
|
||
|
||
/// <summary>
|
||
/// Phase-D server-side HistoryRead continuation-point paging for the Raw arm. Boots a real
|
||
/// <see cref="OtOpcUaSdkServer"/> (as <see cref="NodeManagerHistoryReadTests"/> does), injects an
|
||
/// <see cref="InMemoryHistoryContinuationStore"/> (the in-process harness uses a SESSION-LESS
|
||
/// <c>OperationContext</c>, so the production session-backed store has no session to bind to — the
|
||
/// in-memory store exercises the SAME dispatch path), and a series-backed fake historian, then drives
|
||
/// the full multi-page round trip through the node manager's public HistoryRead.
|
||
/// </summary>
|
||
public sealed class NodeManagerHistoryReadPagingTests : IDisposable
|
||
{
|
||
private static CancellationToken Ct => TestContext.Current.CancellationToken;
|
||
|
||
private readonly string _pkiRoot = Path.Combine(
|
||
Path.GetTempPath(), $"otopcua-historypaging-{Guid.NewGuid():N}");
|
||
|
||
private static readonly DateTime Epoch = new(2026, 1, 1, 0, 0, 0, DateTimeKind.Utc);
|
||
|
||
/// <summary>A full first page emits a continuation point; resuming returns subsequent pages; the
|
||
/// final short page clears the continuation point — and the union of pages is the complete series with
|
||
/// no duplicate or skipped sample. (Resumed pages net slightly fewer than the cap because each resume
|
||
/// re-reads + trims the boundary sample to stay tie-safe — see the design; the union is what matters.)</summary>
|
||
[Fact]
|
||
public async Task Raw_pages_full_series_across_continuation_points()
|
||
{
|
||
var (host, server) = await BootAsync();
|
||
var nm = server.NodeManager!;
|
||
nm.HistoryContinuationStore = new InMemoryHistoryContinuationStore();
|
||
|
||
// 250 distinct-timestamp samples; client caps each page at 100.
|
||
var series = MakeSeries(count: 250, stepSeconds: 1);
|
||
nm.HistorianDataSource = new SeriesHistorianDataSource(series);
|
||
|
||
nm.EnsureVariable("eq-1/pv", parentFolderNodeId: null, displayName: "PV", dataType: "Double",
|
||
writable: false, historianTagname: "WW.PV", realm: AddressSpaceRealm.Uns);
|
||
var nodeId = nm.TryGetVariable("eq-1/pv")!.NodeId;
|
||
|
||
var collected = new List<double>();
|
||
byte[]? cp = null;
|
||
var pageCount = 0;
|
||
var sawCp = false;
|
||
do
|
||
{
|
||
var (values, error, nextCp) = ReadRaw(nm, nodeId, start: Epoch, end: Epoch.AddHours(1),
|
||
max: 100, inboundCp: cp);
|
||
error.StatusCode.Code.ShouldBe(StatusCodes.Good);
|
||
collected.AddRange(values);
|
||
cp = nextCp;
|
||
if (cp is not null) sawCp = true;
|
||
pageCount++;
|
||
pageCount.ShouldBeLessThan(20, "paging must terminate, not loop");
|
||
}
|
||
while (cp is not null);
|
||
|
||
sawCp.ShouldBeTrue("a 250-sample series capped at 100 must page");
|
||
pageCount.ShouldBeGreaterThan(1);
|
||
|
||
// The union is the exact series — no duplicates, no skips, in order.
|
||
collected.Count.ShouldBe(250);
|
||
collected.ShouldBe(Enumerable.Range(0, 250).Select(i => (double)i));
|
||
|
||
await host.DisposeAsync();
|
||
}
|
||
|
||
/// <summary>Part 11: only a StartTime (EndTime unset = MinValue) with a cap pages FORWARD to an open upper bound —
|
||
/// it must not be mistaken for a reverse read, and the stored cursor must not carry MinValue as its end.</summary>
|
||
[Fact]
|
||
public async Task Raw_open_ended_forward_read_pages_ascending_exactly_once()
|
||
{
|
||
var (host, server) = await BootAsync();
|
||
var nm = server.NodeManager!;
|
||
nm.HistoryContinuationStore = new InMemoryHistoryContinuationStore();
|
||
nm.HistorianDataSource = new SeriesHistorianDataSource(MakeSeries(count: 250, stepSeconds: 1));
|
||
nm.EnsureVariable("eq-1/open", parentFolderNodeId: null, displayName: "Open", dataType: "Double",
|
||
writable: false, historianTagname: "WW.Open", realm: AddressSpaceRealm.Uns);
|
||
var nodeId = nm.TryGetVariable("eq-1/open")!.NodeId;
|
||
|
||
var collected = new List<double>();
|
||
byte[]? cp = null;
|
||
var pages = 0;
|
||
do
|
||
{
|
||
var (values, error, next) = ReadRaw(nm, nodeId, Epoch, DateTime.MinValue, max: 100, inboundCp: cp);
|
||
error.StatusCode.Code.ShouldBe(StatusCodes.Good);
|
||
collected.AddRange(values);
|
||
cp = next;
|
||
(++pages).ShouldBeLessThan(20);
|
||
}
|
||
while (cp is not null);
|
||
|
||
collected.ShouldBe(Enumerable.Range(0, 250).Select(i => (double)i));
|
||
await host.DisposeAsync();
|
||
}
|
||
|
||
/// <summary>Part 11: only an EndTime (StartTime unset = MinValue) with a cap reads BACKWARD from the EndTime.</summary>
|
||
[Fact]
|
||
public async Task Raw_end_only_read_pages_descending_from_the_end_time()
|
||
{
|
||
var (host, server) = await BootAsync();
|
||
var nm = server.NodeManager!;
|
||
nm.HistoryContinuationStore = new InMemoryHistoryContinuationStore();
|
||
nm.HistorianDataSource = new SeriesHistorianDataSource(MakeSeries(count: 250, stepSeconds: 1));
|
||
nm.EnsureVariable("eq-1/endonly", parentFolderNodeId: null, displayName: "EndOnly", dataType: "Double",
|
||
writable: false, historianTagname: "WW.EndOnly", realm: AddressSpaceRealm.Uns);
|
||
var nodeId = nm.TryGetVariable("eq-1/endonly")!.NodeId;
|
||
|
||
var collected = new List<double>();
|
||
byte[]? cp = null;
|
||
var pages = 0;
|
||
do
|
||
{
|
||
var (values, error, next) = ReadRaw(nm, nodeId, DateTime.MinValue, Epoch.AddHours(1), max: 100, inboundCp: cp);
|
||
error.StatusCode.Code.ShouldBe(StatusCodes.Good);
|
||
collected.AddRange(values);
|
||
cp = next;
|
||
(++pages).ShouldBeLessThan(20);
|
||
}
|
||
while (cp is not null);
|
||
|
||
collected.ShouldBe(Enumerable.Range(0, 250).Reverse().Select(i => (double)i));
|
||
await host.DisposeAsync();
|
||
}
|
||
|
||
/// <summary>An exactly-full series whose total is a multiple of the cap pages to a trailing
|
||
/// EMPTY page (GoodNoData, no CP) — the server can't know the prior page was the last until it reads
|
||
/// past the end.</summary>
|
||
[Fact]
|
||
public async Task Raw_exact_multiple_terminates_with_empty_final_page()
|
||
{
|
||
var (host, server) = await BootAsync();
|
||
var nm = server.NodeManager!;
|
||
nm.HistoryContinuationStore = new InMemoryHistoryContinuationStore();
|
||
nm.HistorianDataSource = new SeriesHistorianDataSource(MakeSeries(count: 100, stepSeconds: 1));
|
||
|
||
nm.EnsureVariable("eq-1/exact", parentFolderNodeId: null, displayName: "Exact", dataType: "Double",
|
||
writable: false, historianTagname: "WW.Exact", realm: AddressSpaceRealm.Uns);
|
||
var nodeId = nm.TryGetVariable("eq-1/exact")!.NodeId;
|
||
|
||
var (r1, _, cp1) = ReadRaw(nm, nodeId, Epoch, Epoch.AddHours(1), max: 100, inboundCp: null);
|
||
r1.Count.ShouldBe(100);
|
||
cp1.ShouldNotBeNull("a full page emits a CP even when it happens to be the last");
|
||
|
||
var (r2, e2, cp2) = ReadRaw(nm, nodeId, Epoch, Epoch.AddHours(1), max: 100, inboundCp: cp1);
|
||
r2.Count.ShouldBe(0);
|
||
e2.StatusCode.Code.ShouldBe(StatusCodes.Good);
|
||
cp2.ShouldBeNull("the empty trailing page terminates paging");
|
||
|
||
await host.DisposeAsync();
|
||
}
|
||
|
||
/// <summary>Boundary timestamps that TIE across the page split are neither duplicated nor skipped:
|
||
/// the resume cursor drops the already-emitted ties and re-reads the un-emitted ones at the same
|
||
/// timestamp.</summary>
|
||
[Fact]
|
||
public async Task Raw_paging_dedups_tied_boundary_timestamps()
|
||
{
|
||
var (host, server) = await BootAsync();
|
||
var nm = server.NodeManager!;
|
||
nm.HistoryContinuationStore = new InMemoryHistoryContinuationStore();
|
||
|
||
// 5 samples: indices 2,3,4 all share the SAME timestamp (Epoch+2s). With cap 4, page 1 returns
|
||
// [0,1,2,3] (cutting through the tie cluster); page 2 must return [4] (the un-emitted tie) only.
|
||
var ts = new[]
|
||
{
|
||
Epoch, // 0
|
||
Epoch.AddSeconds(1), // 1
|
||
Epoch.AddSeconds(2), // 2 ┐
|
||
Epoch.AddSeconds(2), // 3 │ tied at Epoch+2s
|
||
Epoch.AddSeconds(2), // 4 ┘
|
||
};
|
||
var series = ts.Select((t, i) => new DataValueSnapshot((double)i, StatusCodes.Good, t, t)).ToArray();
|
||
nm.HistorianDataSource = new SeriesHistorianDataSource(series);
|
||
|
||
nm.EnsureVariable("eq-1/tie", parentFolderNodeId: null, displayName: "Tie", dataType: "Double",
|
||
writable: false, historianTagname: "WW.Tie", realm: AddressSpaceRealm.Uns);
|
||
var nodeId = nm.TryGetVariable("eq-1/tie")!.NodeId;
|
||
|
||
var collected = new List<double>();
|
||
|
||
var (r1, _, cp1) = ReadRaw(nm, nodeId, Epoch, Epoch.AddHours(1), max: 4, inboundCp: null);
|
||
r1.Count.ShouldBe(4);
|
||
r1.ShouldBe(new[] { 0.0, 1.0, 2.0, 3.0 });
|
||
cp1.ShouldNotBeNull();
|
||
collected.AddRange(r1);
|
||
|
||
var (r2, _, cp2) = ReadRaw(nm, nodeId, Epoch, Epoch.AddHours(1), max: 4, inboundCp: cp1);
|
||
// Only the un-emitted tie (index 4) — NOT the already-emitted 2 and 3.
|
||
r2.ShouldBe(new[] { 4.0 });
|
||
cp2.ShouldBeNull();
|
||
collected.AddRange(r2);
|
||
|
||
// Exactly the 5 distinct samples, once each.
|
||
collected.ShouldBe(new[] { 0.0, 1.0, 2.0, 3.0, 4.0 });
|
||
|
||
await host.DisposeAsync();
|
||
}
|
||
|
||
/// <summary>Oversized tie cluster (more ties at one timestamp than the page cap): the (timestamp, skip)
|
||
/// resume cursor alone cannot advance past it — the fixed-(start,end,cap) backend keeps returning the
|
||
/// same first <c>cap</c> ties. The paging now over-fetches the WHOLE cluster (a <c>start == end</c> read,
|
||
/// bounded by <c>MaxTieClusterOverfetch</c>) and pages WITHIN the timestamp, so the read drains the
|
||
/// cluster (and any data after it) across continuation points with no dup/skip. Here a tie cluster of 3
|
||
/// (indices 3..7 share one timestamp) sits between distinct samples; with cap 2 the union must be all 10
|
||
/// values in order.</summary>
|
||
[Fact]
|
||
public async Task Raw_oversized_tie_cluster_pages_within_the_timestamp()
|
||
{
|
||
var (host, server) = await BootAsync();
|
||
var nm = server.NodeManager!;
|
||
nm.HistoryContinuationStore = new InMemoryHistoryContinuationStore();
|
||
|
||
// 3 distinct, then a tie cluster of 5 (all at Epoch+3s), then 2 more distinct — 10 total. The tie
|
||
// cluster (5) is larger than the page cap (2), the case that used to stall the cursor.
|
||
var series = MakeSeriesWithTieCluster(distinctBefore: 3, tieCount: 5, distinctAfter: 2);
|
||
series.Length.ShouldBe(10);
|
||
nm.HistorianDataSource = new SeriesHistorianDataSource(series);
|
||
|
||
nm.EnsureVariable("eq-1/burst", parentFolderNodeId: null, displayName: "Burst", dataType: "Double",
|
||
writable: false, historianTagname: "WW.Burst", realm: AddressSpaceRealm.Uns);
|
||
var nodeId = nm.TryGetVariable("eq-1/burst")!.NodeId;
|
||
|
||
var collected = new List<double>();
|
||
byte[]? cp = null;
|
||
var pageCount = 0;
|
||
do
|
||
{
|
||
var (values, error, nextCp) = ReadRaw(nm, nodeId, Epoch, Epoch.AddHours(1),
|
||
max: 2, inboundCp: cp);
|
||
error.StatusCode.Code.ShouldBe(StatusCodes.Good);
|
||
collected.AddRange(values);
|
||
cp = nextCp;
|
||
pageCount++;
|
||
pageCount.ShouldBeLessThan(20, "paging must terminate, not loop, even through a tie cluster");
|
||
}
|
||
while (cp is not null);
|
||
|
||
// Every value, once, in order — the cluster was paged through, not dropped or duplicated.
|
||
collected.Count.ShouldBe(10);
|
||
collected.ShouldBe(Enumerable.Range(0, 10).Select(i => (double)i));
|
||
|
||
await host.DisposeAsync();
|
||
}
|
||
|
||
/// <summary>The absurd-burst backstop is preserved: a tie cluster STRICTLY larger than the configured
|
||
/// <see cref="OtOpcUaNodeManager.MaxTieClusterOverfetch"/> still fails that node rather than buffering an unbounded
|
||
/// burst — <c>BadTooManyOperations</c>, the request-limit code the reverse walk-back uses too (#686; was
|
||
/// <c>BadHistoryOperationUnsupported</c>). With
|
||
/// the bound set to 3 and a 5-way tie cluster, the resume read that hits the cluster fails loudly.</summary>
|
||
[Fact]
|
||
public async Task Raw_tie_cluster_beyond_overfetch_bound_fails_loudly()
|
||
{
|
||
var (host, server) = await BootAsync();
|
||
var nm = server.NodeManager!;
|
||
nm.HistoryContinuationStore = new InMemoryHistoryContinuationStore();
|
||
nm.MaxTieClusterOverfetch = 3; // cluster of 5 will exceed this ⇒ backstop.
|
||
|
||
// 5 samples ALL sharing one timestamp (Epoch+2s) — a tie cluster of 5 with a page cap of 2.
|
||
var t = Epoch.AddSeconds(2);
|
||
var series = Enumerable.Range(0, 5)
|
||
.Select(i => new DataValueSnapshot((double)i, StatusCodes.Good, t, t)).ToArray();
|
||
nm.HistorianDataSource = new SeriesHistorianDataSource(series);
|
||
|
||
nm.EnsureVariable("eq-1/absurd", parentFolderNodeId: null, displayName: "Absurd", dataType: "Double",
|
||
writable: false, historianTagname: "WW.Absurd", realm: AddressSpaceRealm.Uns);
|
||
var nodeId = nm.TryGetVariable("eq-1/absurd")!.NodeId;
|
||
|
||
// Page 1: a full page of the first 2 ties, with a continuation point.
|
||
var (r1, e1, cp1) = ReadRaw(nm, nodeId, Epoch, Epoch.AddHours(1), max: 2, inboundCp: null);
|
||
e1.StatusCode.Code.ShouldBe(StatusCodes.Good);
|
||
r1.ShouldBe(new[] { 0.0, 1.0 });
|
||
cp1.ShouldNotBeNull();
|
||
|
||
// Page 2: the resume read hits the cluster; over-fetch finds 5 > bound 3 ⇒ a clear error, NOT a
|
||
// silent GoodNoData that would drop the un-emitted ties.
|
||
var (r2, e2, cp2) = ReadRaw(nm, nodeId, Epoch, Epoch.AddHours(1), max: 2, inboundCp: cp1);
|
||
e2.StatusCode.Code.ShouldBe(StatusCodes.BadTooManyOperations);
|
||
r2.ShouldBeEmpty();
|
||
cp2.ShouldBeNull();
|
||
|
||
await host.DisposeAsync();
|
||
}
|
||
|
||
/// <summary>NumValuesPerNode == 0 ("all values") never pages — the whole series returns in one shot
|
||
/// with a null continuation point.</summary>
|
||
[Fact]
|
||
public async Task Raw_unlimited_request_returns_everything_without_a_cp()
|
||
{
|
||
var (host, server) = await BootAsync();
|
||
var nm = server.NodeManager!;
|
||
nm.HistoryContinuationStore = new InMemoryHistoryContinuationStore();
|
||
nm.HistorianDataSource = new SeriesHistorianDataSource(MakeSeries(count: 250, stepSeconds: 1));
|
||
|
||
nm.EnsureVariable("eq-1/all", parentFolderNodeId: null, displayName: "All", dataType: "Double",
|
||
writable: false, historianTagname: "WW.All", realm: AddressSpaceRealm.Uns);
|
||
var nodeId = nm.TryGetVariable("eq-1/all")!.NodeId;
|
||
|
||
var (r, e, cp) = ReadRaw(nm, nodeId, Epoch, Epoch.AddHours(1), max: 0, inboundCp: null);
|
||
|
||
e.StatusCode.Code.ShouldBe(StatusCodes.Good);
|
||
r.Count.ShouldBe(250);
|
||
cp.ShouldBeNull("an unlimited (max==0) request must not page");
|
||
|
||
await host.DisposeAsync();
|
||
}
|
||
|
||
/// <summary>An inbound continuation point the store doesn't know (released / never issued / from
|
||
/// another node) yields BadContinuationPointInvalid for that node, and the source is NOT read.</summary>
|
||
[Fact]
|
||
public async Task Raw_unknown_continuation_point_yields_BadContinuationPointInvalid()
|
||
{
|
||
var (host, server) = await BootAsync();
|
||
var nm = server.NodeManager!;
|
||
nm.HistoryContinuationStore = new InMemoryHistoryContinuationStore();
|
||
var fake = new SeriesHistorianDataSource(MakeSeries(count: 10, stepSeconds: 1));
|
||
nm.HistorianDataSource = fake;
|
||
|
||
nm.EnsureVariable("eq-1/bad-cp", parentFolderNodeId: null, displayName: "BadCp", dataType: "Double",
|
||
writable: false, historianTagname: "WW.BadCp", realm: AddressSpaceRealm.Uns);
|
||
var nodeId = nm.TryGetVariable("eq-1/bad-cp")!.NodeId;
|
||
|
||
fake.ResetReadCount();
|
||
var bogus = Guid.NewGuid().ToByteArray();
|
||
var (_, e, cp) = ReadRaw(nm, nodeId, Epoch, Epoch.AddHours(1), max: 100, inboundCp: bogus);
|
||
|
||
e.StatusCode.Code.ShouldBe(StatusCodes.BadContinuationPointInvalid);
|
||
cp.ShouldBeNull();
|
||
fake.ReadCount.ShouldBe(0); // never reached the backend.
|
||
|
||
await host.DisposeAsync();
|
||
}
|
||
|
||
/// <summary>releaseContinuationPoints drops the stored cursor WITHOUT reading data: a subsequent
|
||
/// resume of the released point then misses (BadContinuationPointInvalid).</summary>
|
||
[Fact]
|
||
public async Task Release_drops_the_cursor_without_reading()
|
||
{
|
||
var (host, server) = await BootAsync();
|
||
var nm = server.NodeManager!;
|
||
nm.HistoryContinuationStore = new InMemoryHistoryContinuationStore();
|
||
var fake = new SeriesHistorianDataSource(MakeSeries(count: 250, stepSeconds: 1));
|
||
nm.HistorianDataSource = fake;
|
||
|
||
nm.EnsureVariable("eq-1/rel", parentFolderNodeId: null, displayName: "Rel", dataType: "Double",
|
||
writable: false, historianTagname: "WW.Rel", realm: AddressSpaceRealm.Uns);
|
||
var nodeId = nm.TryGetVariable("eq-1/rel")!.NodeId;
|
||
|
||
// Page 1 — get a CP.
|
||
var (_, _, cp1) = ReadRaw(nm, nodeId, Epoch, Epoch.AddHours(1), max: 100, inboundCp: null);
|
||
cp1.ShouldNotBeNull();
|
||
|
||
// Release it — the dispatcher routes releaseContinuationPoints=true to HistoryReleaseContinuationPoints,
|
||
// which never reaches the Raw arm. No data is read.
|
||
fake.ResetReadCount();
|
||
InvokeRelease(nm, nodeId, cp1!);
|
||
fake.ReadCount.ShouldBe(0);
|
||
|
||
// Resuming the released point now misses.
|
||
var (_, e, _) = ReadRaw(nm, nodeId, Epoch, Epoch.AddHours(1), max: 100, inboundCp: cp1);
|
||
e.StatusCode.Code.ShouldBe(StatusCodes.BadContinuationPointInvalid);
|
||
|
||
await host.DisposeAsync();
|
||
}
|
||
|
||
/// <summary>
|
||
/// A reverse read (start > end) with <c>NumValuesPerNode = 1</c> over three distinct-timestamp samples must
|
||
/// page through every sample, newest first. Before the fix the second page's resume read re-returned the
|
||
/// boundary sample, the trim emptied it, the tie-cluster path drained the one-sample "cluster" and stepped
|
||
/// FORWARD to <c>T + 1 tick</c> — past a reverse window's end — so paging stopped after the first sample.
|
||
/// </summary>
|
||
[Fact]
|
||
public async Task Raw_reverse_read_with_one_value_per_page_returns_every_sample_descending()
|
||
{
|
||
var (host, server) = await BootAsync();
|
||
var nm = server.NodeManager!;
|
||
nm.HistoryContinuationStore = new InMemoryHistoryContinuationStore();
|
||
nm.HistorianDataSource = new SeriesHistorianDataSource(MakeSeries(count: 3, stepSeconds: 1));
|
||
|
||
nm.EnsureVariable("eq-1/rev", parentFolderNodeId: null, displayName: "Rev", dataType: "Double",
|
||
writable: false, historianTagname: "WW.Rev", realm: AddressSpaceRealm.Uns);
|
||
var nodeId = nm.TryGetVariable("eq-1/rev")!.NodeId;
|
||
|
||
var collected = await DrainAsync(nm, nodeId, start: Epoch.AddHours(1), end: Epoch.AddSeconds(-1), max: 1);
|
||
|
||
collected.ShouldBe([2d, 1d, 0d]);
|
||
await host.DisposeAsync();
|
||
}
|
||
|
||
/// <summary>The reverse mirror of <see cref="Raw_oversized_tie_cluster_pages_within_the_timestamp"/>: a 5-way tie
|
||
/// cluster between distinct samples, cap 2, read newest-first — every value once, in descending order.</summary>
|
||
[Fact]
|
||
public async Task Raw_reverse_read_pages_through_an_oversized_tie_cluster()
|
||
{
|
||
var (host, server) = await BootAsync();
|
||
var nm = server.NodeManager!;
|
||
nm.HistoryContinuationStore = new InMemoryHistoryContinuationStore();
|
||
nm.HistorianDataSource = new SeriesHistorianDataSource(
|
||
MakeSeriesWithTieCluster(distinctBefore: 3, tieCount: 5, distinctAfter: 2));
|
||
|
||
nm.EnsureVariable("eq-1/revburst", parentFolderNodeId: null, displayName: "RevBurst", dataType: "Double",
|
||
writable: false, historianTagname: "WW.RevBurst", realm: AddressSpaceRealm.Uns);
|
||
var nodeId = nm.TryGetVariable("eq-1/revburst")!.NodeId;
|
||
|
||
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));
|
||
await host.DisposeAsync();
|
||
}
|
||
|
||
/// <summary>
|
||
/// Forward paging at <c>NumValuesPerNode = 1</c> over a MICROSECOND backend (QuestDB: it truncates every inbound
|
||
/// time to 10 ticks and reports <c>TimestampResolutionTicks = 10</c>) returns every sample exactly once and
|
||
/// terminates. Before the fix the drained one-sample "cluster" resumed at <c>T + 1 tick</c>, which the backend
|
||
/// truncated back to <c>T</c>: the same sample came back with no boundary skip, and the read looped forever.
|
||
/// </summary>
|
||
[Fact]
|
||
public async Task Raw_forward_read_with_one_value_per_page_on_a_microsecond_backend_returns_every_sample_once()
|
||
{
|
||
var (host, server) = await BootAsync();
|
||
var nm = server.NodeManager!;
|
||
nm.HistoryContinuationStore = new InMemoryHistoryContinuationStore();
|
||
// Microsecond-spaced samples, so a one-tick step would land strictly between two of them.
|
||
var series = Enumerable.Range(0, 6)
|
||
.Select(i => { var t = Epoch.AddTicks(10L * i); return new DataValueSnapshot((double)i, StatusCodes.Good, t, t); })
|
||
.ToArray();
|
||
nm.HistorianDataSource = new MicrosecondHistorianDataSource(series);
|
||
|
||
nm.EnsureVariable("eq-1/us", parentFolderNodeId: null, displayName: "Us", dataType: "Double",
|
||
writable: false, historianTagname: "WW.Us", realm: AddressSpaceRealm.Uns);
|
||
var nodeId = nm.TryGetVariable("eq-1/us")!.NodeId;
|
||
|
||
var collected = await DrainAsync(nm, nodeId, start: Epoch, end: Epoch.AddHours(1), max: 1);
|
||
|
||
collected.ShouldBe([0d, 1d, 2d, 3d, 4d, 5d]);
|
||
await host.DisposeAsync();
|
||
}
|
||
|
||
/// <summary>The reverse mirror on the same microsecond backend: newest first, each sample once.</summary>
|
||
[Fact]
|
||
public async Task Raw_reverse_read_with_one_value_per_page_on_a_microsecond_backend_returns_every_sample_once()
|
||
{
|
||
var (host, server) = await BootAsync();
|
||
var nm = server.NodeManager!;
|
||
nm.HistoryContinuationStore = new InMemoryHistoryContinuationStore();
|
||
var series = Enumerable.Range(0, 6)
|
||
.Select(i => { var t = Epoch.AddTicks(10L * (i + 1)); return new DataValueSnapshot((double)i, StatusCodes.Good, t, t); })
|
||
.ToArray();
|
||
nm.HistorianDataSource = new MicrosecondHistorianDataSource(series);
|
||
|
||
nm.EnsureVariable("eq-1/usrev", parentFolderNodeId: null, displayName: "UsRev", dataType: "Double",
|
||
writable: false, historianTagname: "WW.UsRev", realm: AddressSpaceRealm.Uns);
|
||
var nodeId = nm.TryGetVariable("eq-1/usrev")!.NodeId;
|
||
|
||
var collected = await DrainAsync(nm, nodeId, start: Epoch.AddHours(1), end: Epoch, max: 1);
|
||
|
||
collected.ShouldBe([5d, 4d, 3d, 2d, 1d, 0d]);
|
||
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 > 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);
|
||
source.ReadCount.ShouldBe(1, "an uncapped reverse read is ONE forward read, reversed in memory — not a walk");
|
||
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(2 * 16 + 4);
|
||
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.ShouldBeLessThanOrEqualTo(20);
|
||
source.MinRequestedStart.ShouldBeGreaterThanOrEqualTo(DateTime.UnixEpoch, "the walk never asks for data before 1970");
|
||
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>#686 review: a dense tag is not re-discovered every page — the width the walk ended on rides the
|
||
/// continuation point, so after the first page each page is about one read, with no full (discarded) reads.</summary>
|
||
[Fact]
|
||
public async Task Raw_reverse_paging_on_a_forward_only_backend_carries_the_observed_density_between_pages()
|
||
{
|
||
// 2000 samples 10 ms apart: a 1-minute first segment holds all of them, far over a budget of 64.
|
||
var series = Enumerable.Range(0, 2000)
|
||
.Select(i => { var t = Epoch.AddMilliseconds(10L * i); return new DataValueSnapshot((double)i, StatusCodes.Good, t, t); })
|
||
.ToArray();
|
||
var source = new ForwardOnlyHistorianDataSource(series);
|
||
var (host, nm, nodeId) = await BootForwardOnlyAsync(source, "FwdDensity");
|
||
nm.MaxTieClusterOverfetch = 64;
|
||
|
||
// Page 1 finds the data (walking down an empty hour) and its density; measure the pages after it.
|
||
var (first, _, cp) = ReadRaw(nm, nodeId, DateTime.MinValue, Epoch.AddSeconds(20), max: 20, inboundCp: null);
|
||
first.ShouldBe(Enumerable.Range(1980, 20).Reverse().Select(i => (double)i));
|
||
source.ResetCounts();
|
||
|
||
var collected = new List<double>(first);
|
||
var pages = 0;
|
||
while (cp is not null)
|
||
{
|
||
var (values, error, next) = ReadRaw(nm, nodeId, DateTime.MinValue, Epoch.AddSeconds(20), max: 20, inboundCp: cp);
|
||
error.StatusCode.Code.ShouldBe(StatusCodes.Good);
|
||
collected.AddRange(values);
|
||
cp = next;
|
||
(++pages).ShouldBeLessThan(200);
|
||
}
|
||
|
||
collected.ShouldBe(Enumerable.Range(0, 2000).Reverse().Select(i => (double)i));
|
||
// The discriminating count: reads that came back full and were thrown away. Without the carried width every page
|
||
// starts on a 1-minute segment holding all 2000 samples and discards full reads first (~2 per page).
|
||
source.FullReads.ShouldBeLessThanOrEqualTo(2);
|
||
// About one read per page, plus the final pages' walks from the first sample down to the 1970 floor (~16 each).
|
||
source.ReadCount.ShouldBeLessThanOrEqualTo(pages + 40);
|
||
await host.DisposeAsync();
|
||
}
|
||
|
||
/// <summary>
|
||
/// #689: a paged end-only read on a dense tag sizes each segment from the PAGE (N), not from
|
||
/// <c>MaxTieClusterOverfetch</c>. Live (AVEVA, 1 Hz, N = 3): from about page 5 every page read 32,770 rows and took
|
||
/// 2–5 s to return 2 rows, because the walk re-aimed its width at half the 65,536-row tie budget and carried it.
|
||
/// </summary>
|
||
[Fact]
|
||
public async Task Raw_reverse_paging_on_a_dense_forward_only_tag_reads_about_the_page_size_per_page()
|
||
{
|
||
// 100,000 samples 100 ms apart (~2.8 h): far more than the default tie budget would ever let a page read.
|
||
var series = Enumerable.Range(0, 100_000)
|
||
.Select(i => { var t = Epoch.AddMilliseconds(100L * i); return new DataValueSnapshot((double)i, StatusCodes.Good, t, t); })
|
||
.ToArray();
|
||
var source = new ForwardOnlyHistorianDataSource(series);
|
||
var (host, nm, nodeId) = await BootForwardOnlyAsync(source, "FwdDense689");
|
||
nm.MaxTieClusterOverfetch.ShouldBe(65536, "the default budget — the one the live gate ran with");
|
||
var end = Epoch.AddMilliseconds(100L * 99_999);
|
||
|
||
byte[]? cp = null;
|
||
var expected = 99_999;
|
||
for (var page = 1; page <= 20; page++)
|
||
{
|
||
source.ResetCounts();
|
||
var (values, error, next) = ReadRaw(nm, nodeId, DateTime.MinValue, end, max: 3, inboundCp: cp);
|
||
error.StatusCode.Code.ShouldBe(StatusCodes.Good);
|
||
values.ShouldNotBeEmpty();
|
||
values[^1].ShouldBe((double)(expected - 2), $"page {page} continues exactly");
|
||
expected = (int)values[^1];
|
||
next.ShouldNotBeNull();
|
||
cp = next;
|
||
|
||
// The discriminating counts: rows read and the largest cap asked for, per page. Page 1 has no density yet and
|
||
// may read one discovery segment (a minute here, 600 rows); every page after it is sized from N.
|
||
var rowsBound = page == 1 ? ReverseWalk.DiscoveryRows + 2 : 64;
|
||
source.RowsReturned.ShouldBeLessThanOrEqualTo(rowsBound, $"page {page} read {source.RowsReturned} rows to return {values.Count}");
|
||
source.MaxCapRequested.ShouldBeLessThanOrEqualTo((uint)rowsBound, $"page {page} asked for {source.MaxCapRequested} rows");
|
||
source.ReadCount.ShouldBeLessThanOrEqualTo(3, $"page {page}");
|
||
}
|
||
|
||
await host.DisposeAsync();
|
||
}
|
||
|
||
/// <summary>
|
||
/// #689/#691: randomized end-to-end paging of end-only reads through the node manager on a forward-only backend —
|
||
/// bursts, gaps, sub-millisecond stamps and tie clusters (some larger than the page) — the union of the pages is
|
||
/// every sample exactly once, newest first, whatever N and the tie budget.
|
||
/// </summary>
|
||
[Fact]
|
||
public async Task Raw_reverse_paging_on_a_forward_only_backend_is_exact_on_random_data()
|
||
{
|
||
var (host, nm, nodeId) = await BootForwardOnlyAsync(new ForwardOnlyHistorianDataSource([]), "FwdFuzz");
|
||
for (var seed = 0; seed < 150; seed++)
|
||
{
|
||
var rng = new Random(seed);
|
||
var series = new List<DataValueSnapshot>();
|
||
var t = Epoch.Ticks;
|
||
var v = 0;
|
||
var count = rng.Next(1, 300);
|
||
while (series.Count < count)
|
||
{
|
||
t += rng.Next(5) switch
|
||
{
|
||
0 => rng.Next(1, 10),
|
||
1 => rng.Next(1, 5) * TimeSpan.TicksPerMillisecond,
|
||
2 => rng.Next(1, 90) * TimeSpan.TicksPerMinute,
|
||
_ => rng.Next(1, 2000) * TimeSpan.TicksPerMillisecond,
|
||
};
|
||
var ties = rng.Next(6) == 0 ? rng.Next(2, 7) : 1;
|
||
for (var k = 0; k < ties; k++)
|
||
{
|
||
var ts = new DateTime(t, DateTimeKind.Utc);
|
||
series.Add(new DataValueSnapshot((double)v++, StatusCodes.Good, ts, ts));
|
||
}
|
||
}
|
||
|
||
// No fabricated start row: the Gateway source drops it (#687) — the node manager's tie-cluster point read relies on
|
||
// that. The walk itself is fuzzed WITH the row in ReverseWalkTests.
|
||
nm.HistorianDataSource = new ForwardOnlyHistorianDataSource(series, emitBoundaryRow: false);
|
||
nm.MaxTieClusterOverfetch = rng.Next(2) == 0 ? 8 : 65536;
|
||
var max = (uint)rng.Next(1, 12);
|
||
var end = new DateTime(t + rng.Next(0, 3) * TimeSpan.TicksPerSecond, DateTimeKind.Utc);
|
||
|
||
// The walk refuses more than MaxTieClusterOverfetch samples within ONE MILLISECOND (#691.4 re-review), so with
|
||
// the small budget a dense sub-millisecond burst is a legitimate BadTooManyOperations — and nothing else is.
|
||
var msMax = 0;
|
||
for (int i = 0, j = 0; i < series.Count; i++)
|
||
{
|
||
while (series[i].SourceTimestampUtc!.Value.Ticks - series[j].SourceTimestampUtc!.Value.Ticks >= TimeSpan.TicksPerMillisecond) j++;
|
||
msMax = Math.Max(msMax, i - j + 1);
|
||
}
|
||
|
||
var collected = new List<double>();
|
||
byte[]? cp = null;
|
||
var pages = 0;
|
||
var refused = false;
|
||
do
|
||
{
|
||
var (values, error, next) = ReadRaw(nm, nodeId, DateTime.MinValue, end, max, inboundCp: cp);
|
||
if (error.StatusCode.Code == StatusCodes.BadTooManyOperations && msMax > nm.MaxTieClusterOverfetch)
|
||
{
|
||
refused = true;
|
||
break;
|
||
}
|
||
|
||
error.StatusCode.Code.ShouldBe(StatusCodes.Good, $"seed {seed}, N = {max}, ms max {msMax}");
|
||
collected.AddRange(values);
|
||
cp = next;
|
||
(++pages).ShouldBeLessThan(2000, "paging must terminate, not loop");
|
||
}
|
||
while (cp is not null);
|
||
|
||
var expected = Enumerable.Range(0, series.Count).Reverse().Select(i => (double)i).ToList();
|
||
if (refused) collected.ShouldBe(expected.Take(collected.Count), $"seed {seed}: what came before the refusal");
|
||
else collected.ShouldBe(expected, $"seed {seed}, N = {max}");
|
||
}
|
||
|
||
await host.DisposeAsync();
|
||
}
|
||
|
||
/// <summary>
|
||
/// #691: the last pages of a deep reverse page-through no longer each walk down to 1970 (live: ~14–16 reads, ~2.5 s
|
||
/// per page). The stretch a page proved empty below its oldest row rides the continuation point, and a page that runs
|
||
/// out of rows asks once for the whole remainder — so every page here is a couple of reads.
|
||
/// </summary>
|
||
[Fact]
|
||
public async Task Raw_reverse_page_through_on_a_forward_only_backend_does_not_rewalk_to_the_epoch_on_each_last_page()
|
||
{
|
||
var source = new ForwardOnlyHistorianDataSource(MakeSeries(count: 25, stepSeconds: 1));
|
||
var (host, nm, nodeId) = await BootForwardOnlyAsync(source, "FwdDeep691");
|
||
nm.Clock = new FixedClock(Epoch.AddHours(1));
|
||
|
||
var collected = new List<double>();
|
||
byte[]? cp = null;
|
||
var page = 0;
|
||
do
|
||
{
|
||
source.ResetCounts();
|
||
var (values, error, next) = ReadRaw(nm, nodeId, DateTime.MinValue, Epoch.AddHours(1), max: 10, inboundCp: cp);
|
||
error.StatusCode.Code.ShouldBe(StatusCodes.Good);
|
||
collected.AddRange(values);
|
||
cp = next;
|
||
page++;
|
||
source.ReadCount.ShouldBeLessThanOrEqualTo(page == 1 ? 4 : 1, $"page {page}");
|
||
}
|
||
while (cp is not null && page < 10);
|
||
|
||
collected.ShouldBe(Enumerable.Range(0, 25).Reverse().Select(i => (double)i));
|
||
source.MinRequestedStart.ShouldBeGreaterThanOrEqualTo(DateTime.UnixEpoch);
|
||
await host.DisposeAsync();
|
||
}
|
||
|
||
/// <summary>#686 review: a reverse read from <c>StartTime = DateTime.MaxValue</c> walks back from now + 1 day, not from
|
||
/// year 9999.</summary>
|
||
[Fact]
|
||
public async Task Raw_reverse_read_from_max_value_on_a_forward_only_backend_starts_at_now_plus_one_day()
|
||
{
|
||
var source = new ForwardOnlyHistorianDataSource(MakeSeries(count: 250, stepSeconds: 1));
|
||
var (host, nm, nodeId) = await BootForwardOnlyAsync(source, "FwdMax");
|
||
var now = Epoch.AddHours(1);
|
||
nm.Clock = new FixedClock(now);
|
||
|
||
var (values, error, _) = ReadRaw(nm, nodeId, DateTime.MaxValue, Epoch.AddSeconds(-1), max: 3, inboundCp: null);
|
||
|
||
error.StatusCode.Code.ShouldBe(StatusCodes.Good);
|
||
values.ShouldBe([249d, 248d, 247d]);
|
||
source.MaxRequestedEnd.ShouldBeLessThanOrEqualTo(now.AddDays(1).AddMilliseconds(1));
|
||
source.ReadCount.ShouldBeLessThanOrEqualTo(10);
|
||
await host.DisposeAsync();
|
||
}
|
||
|
||
/// <summary>#686 review: AVEVA may stamp the row it fabricates at a query start rounded UP to the millisecond. The walk
|
||
/// requests millisecond-aligned starts, so that row can never land inside (EndTime, …] when EndTime is not on a
|
||
/// millisecond.</summary>
|
||
[Fact]
|
||
public async Task Raw_reverse_read_on_a_forward_only_backend_never_returns_a_boundary_row_rounded_up_into_the_window()
|
||
{
|
||
var source = new ForwardOnlyHistorianDataSource(MakeSeries(count: 250, stepSeconds: 1), boundaryAtCeilingMs: true);
|
||
var (host, nm, nodeId) = await BootForwardOnlyAsync(source, "FwdCeil");
|
||
|
||
var (values, error, _) = ReadRaw(nm, nodeId, Epoch.AddSeconds(10), Epoch.AddTicks(5_000), max: 100, inboundCp: null);
|
||
|
||
error.StatusCode.Code.ShouldBe(StatusCodes.Good);
|
||
values.ShouldBe(Enumerable.Range(1, 10).Reverse().Select(i => (double)i)); // (Epoch+0.5ms, Epoch+10s]: no 0
|
||
await host.DisposeAsync();
|
||
}
|
||
|
||
/// <summary>#686 review: the upper edge is filtered too — the request end is rounded UP to the millisecond, so a
|
||
/// sample one tick after StartTime comes back from the backend and must not be returned.</summary>
|
||
[Fact]
|
||
public async Task Raw_reverse_read_on_a_forward_only_backend_excludes_a_sample_one_tick_after_the_start()
|
||
{
|
||
DataValueSnapshot At(long ticks, double v)
|
||
{
|
||
var t = Epoch.AddTicks(ticks);
|
||
return new DataValueSnapshot(v, StatusCodes.Good, t, t);
|
||
}
|
||
|
||
var t0 = TimeSpan.TicksPerSecond;
|
||
var source = new ForwardOnlyHistorianDataSource([At(0, 0), At(t0, 1), At(t0 + 1, 99)]);
|
||
var (host, nm, nodeId) = await BootForwardOnlyAsync(source, "FwdUpper");
|
||
|
||
var (values, error, _) = ReadRaw(nm, nodeId, Epoch.AddTicks(t0), Epoch.AddSeconds(-1), max: 10, inboundCp: null);
|
||
|
||
error.StatusCode.Code.ShouldBe(StatusCodes.Good);
|
||
values.ShouldBe([1d, 0d]);
|
||
await host.DisposeAsync();
|
||
}
|
||
|
||
/// <summary>#686 review: a tie cluster of EXACTLY <c>MaxTieClusterOverfetch</c> is served — the row the backend
|
||
/// fabricates at the query start does not consume the budget.</summary>
|
||
[Fact]
|
||
public async Task Raw_reverse_read_on_a_forward_only_backend_serves_a_tie_cluster_of_exactly_the_budget()
|
||
{
|
||
var source = new ForwardOnlyHistorianDataSource(MakeSeriesWithTieCluster(distinctBefore: 2, tieCount: 4, distinctAfter: 2));
|
||
var (host, nm, nodeId) = await BootForwardOnlyAsync(source, "FwdExact");
|
||
nm.MaxTieClusterOverfetch = 4;
|
||
|
||
var (values, error, _) = ReadRaw(nm, nodeId, Epoch.AddHours(1), Epoch.AddSeconds(-1), max: 100, inboundCp: null);
|
||
|
||
error.StatusCode.Code.ShouldBe(StatusCodes.Good);
|
||
values.ShouldBe(Enumerable.Range(0, 8).Reverse().Select(i => (double)i));
|
||
await host.DisposeAsync();
|
||
}
|
||
|
||
/// <summary>#686 review: adjacent tie clusters one millisecond apart, none over the budget, are served — a neighbour
|
||
/// inside the request's alignment slack does not get the segment refused.</summary>
|
||
[Fact]
|
||
public async Task Raw_reverse_read_on_a_forward_only_backend_serves_adjacent_tie_clusters_within_the_budget()
|
||
{
|
||
var series = new List<DataValueSnapshot>();
|
||
var v = 0;
|
||
for (var c = 0; c < 3; c++)
|
||
{
|
||
var t = Epoch.AddSeconds(5).AddMilliseconds(c);
|
||
for (var i = 0; i < 4; i++) series.Add(new DataValueSnapshot((double)v++, StatusCodes.Good, t, t));
|
||
}
|
||
|
||
var source = new ForwardOnlyHistorianDataSource(series);
|
||
var (host, nm, nodeId) = await BootForwardOnlyAsync(source, "FwdAdjacent");
|
||
nm.MaxTieClusterOverfetch = 4;
|
||
|
||
var (values, error, _) = ReadRaw(nm, nodeId, Epoch.AddHours(1), Epoch.AddSeconds(-1), max: 100, inboundCp: null);
|
||
|
||
error.StatusCode.Code.ShouldBe(StatusCodes.Good);
|
||
values.ShouldBe(Enumerable.Range(0, 12).Reverse().Select(i => (double)i));
|
||
await host.DisposeAsync();
|
||
}
|
||
|
||
/// <summary>#686 review: a <c>MaxTieClusterOverfetch</c> of 1 (or a misconfigured 0) does not make every reverse read on
|
||
/// a forward-only backend refuse — distinct-timestamp data is still served.</summary>
|
||
[Theory]
|
||
[InlineData(1)]
|
||
[InlineData(0)]
|
||
public async Task Raw_reverse_read_on_a_forward_only_backend_works_with_a_minimal_budget(int budget)
|
||
{
|
||
var source = new ForwardOnlyHistorianDataSource(MakeSeries(count: 250, stepSeconds: 1));
|
||
var (host, nm, nodeId) = await BootForwardOnlyAsync(source, $"FwdMin{budget}");
|
||
nm.MaxTieClusterOverfetch = budget;
|
||
|
||
var (values, error, _) = ReadRaw(nm, nodeId, DateTime.MinValue, Epoch.AddSeconds(100), max: 3, inboundCp: null);
|
||
|
||
error.StatusCode.Code.ShouldBe(StatusCodes.Good);
|
||
values.ShouldBe([100d, 99d, 98d]);
|
||
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,
|
||
int maxPages = 40)
|
||
{
|
||
var collected = new List<double>();
|
||
byte[]? cp = null;
|
||
var pageCount = 0;
|
||
do
|
||
{
|
||
var (values, error, nextCp) = ReadRaw(nm, nodeId, start, end, max, inboundCp: cp);
|
||
error.StatusCode.Code.ShouldBe(StatusCodes.Good);
|
||
collected.AddRange(values);
|
||
cp = nextCp;
|
||
pageCount++;
|
||
pageCount.ShouldBeLessThan(maxPages, "paging must terminate, not loop");
|
||
}
|
||
while (cp is not null);
|
||
return Task.FromResult(collected);
|
||
}
|
||
|
||
// --- helpers --------------------------------------------------------------------------------
|
||
|
||
/// <summary>Issue a single-node Raw HistoryRead and return that node's decoded sample values, error,
|
||
/// and outbound continuation point.</summary>
|
||
private static (IReadOnlyList<double> Values, ServiceResult Error, byte[]? Cp) ReadRaw(
|
||
OtOpcUaNodeManager nm, NodeId nodeId, DateTime start, DateTime end, uint max, byte[]? inboundCp)
|
||
{
|
||
var context = new OperationContext(
|
||
new RequestHeader(), secureChannelContext: null, RequestType.HistoryRead, identity: null);
|
||
|
||
var details = new ReadRawModifiedDetails
|
||
{
|
||
StartTime = start,
|
||
EndTime = end,
|
||
NumValuesPerNode = max,
|
||
IsReadModified = false,
|
||
};
|
||
|
||
var nodesToRead = new List<HistoryReadValueId>
|
||
{
|
||
new() { NodeId = nodeId, ContinuationPoint = inboundCp },
|
||
};
|
||
var results = new List<SdkHistoryReadResult> { null! };
|
||
var errors = new List<ServiceResult> { null! };
|
||
|
||
nm.HistoryRead(context, details, TimestampsToReturn.Both,
|
||
releaseContinuationPoints: false, nodesToRead, results, errors);
|
||
|
||
var values = new List<double>();
|
||
if (results[0]?.HistoryData is { } ho && !ExtensionObject.IsNull(ho))
|
||
{
|
||
var data = (HistoryData)ExtensionObject.ToEncodeable(ho);
|
||
foreach (var dv in data.DataValues) values.Add((double)dv.Value);
|
||
}
|
||
|
||
return (values, errors[0], results[0]?.ContinuationPoint);
|
||
}
|
||
|
||
/// <summary>Issue a release-only HistoryRead (releaseContinuationPoints=true) for the node + point.</summary>
|
||
private static void InvokeRelease(OtOpcUaNodeManager nm, NodeId nodeId, byte[] cp)
|
||
{
|
||
var context = new OperationContext(
|
||
new RequestHeader(), secureChannelContext: null, RequestType.HistoryRead, identity: null);
|
||
var details = new ReadRawModifiedDetails
|
||
{
|
||
StartTime = Epoch, EndTime = Epoch.AddHours(1), NumValuesPerNode = 100, IsReadModified = false,
|
||
};
|
||
var nodesToRead = new List<HistoryReadValueId> { new() { NodeId = nodeId, ContinuationPoint = cp } };
|
||
var results = new List<SdkHistoryReadResult> { null! };
|
||
var errors = new List<ServiceResult> { null! };
|
||
|
||
nm.HistoryRead(context, details, TimestampsToReturn.Both,
|
||
releaseContinuationPoints: true, nodesToRead, results, errors);
|
||
}
|
||
|
||
private static DataValueSnapshot[] MakeSeries(int count, int stepSeconds) =>
|
||
Enumerable.Range(0, count)
|
||
.Select(i =>
|
||
{
|
||
var t = Epoch.AddSeconds((long)i * stepSeconds);
|
||
return new DataValueSnapshot((double)i, StatusCodes.Good, t, t);
|
||
})
|
||
.ToArray();
|
||
|
||
/// <summary>Build a sorted series: <paramref name="distinctBefore"/> distinct-second samples, then a
|
||
/// run of <paramref name="tieCount"/> samples ALL at the same timestamp (the next second after the
|
||
/// last distinct one — the oversized tie cluster), then <paramref name="distinctAfter"/> distinct
|
||
/// samples each a second later. Values are 0..N-1 in series order so the union assertion is exact.</summary>
|
||
private static DataValueSnapshot[] MakeSeriesWithTieCluster(int distinctBefore, int tieCount, int distinctAfter)
|
||
{
|
||
var samples = new List<DataValueSnapshot>(distinctBefore + tieCount + distinctAfter);
|
||
var value = 0;
|
||
var second = 0;
|
||
|
||
for (var i = 0; i < distinctBefore; i++, second++)
|
||
{
|
||
var t = Epoch.AddSeconds(second);
|
||
samples.Add(new DataValueSnapshot((double)value++, StatusCodes.Good, t, t));
|
||
}
|
||
|
||
var tieT = Epoch.AddSeconds(second++); // the single shared timestamp for the whole cluster
|
||
for (var i = 0; i < tieCount; i++)
|
||
samples.Add(new DataValueSnapshot((double)value++, StatusCodes.Good, tieT, tieT));
|
||
|
||
for (var i = 0; i < distinctAfter; i++, second++)
|
||
{
|
||
var t = Epoch.AddSeconds(second);
|
||
samples.Add(new DataValueSnapshot((double)value++, StatusCodes.Good, t, t));
|
||
}
|
||
|
||
return samples.ToArray();
|
||
}
|
||
|
||
/// <summary>A series-backed fake historian: holds a full sorted series and serves a Raw read by
|
||
/// returning the samples in [start, end] capped at maxValuesPerNode (start inclusive — exactly the
|
||
/// resume-cursor contract). Processed / AtTime / Events are not exercised here.</summary>
|
||
private sealed class SeriesHistorianDataSource(IReadOnlyList<DataValueSnapshot> series) : IHistorianDataSource
|
||
{
|
||
private int _readCount;
|
||
|
||
/// <summary>Serves start > 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;
|
||
|
||
public Task<HistorianRead> ReadRawAsync(
|
||
string fullReference, DateTime startUtc, DateTime endUtc, uint maxValuesPerNode,
|
||
CancellationToken cancellationToken)
|
||
{
|
||
Interlocked.Increment(ref _readCount);
|
||
// Reverse (Part 11, start > end): (end, start] in DESCENDING time — the SQL backend's contract.
|
||
// Ties keep series order reversed, so a reverse page of a cluster is deterministic.
|
||
var window = startUtc > endUtc
|
||
? series
|
||
.Where(s => (s.SourceTimestampUtc ?? DateTime.MinValue) > endUtc
|
||
&& (s.SourceTimestampUtc ?? DateTime.MinValue) <= startUtc)
|
||
.Reverse()
|
||
.ToList()
|
||
: series
|
||
.Where(s => (s.SourceTimestampUtc ?? DateTime.MinValue) >= startUtc
|
||
&& (s.SourceTimestampUtc ?? DateTime.MinValue) <= endUtc)
|
||
.ToList();
|
||
// maxValuesPerNode == 0 means "no limit".
|
||
var page = maxValuesPerNode == 0 ? window : window.Take((int)maxValuesPerNode).ToList();
|
||
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>
|
||
/// #686: a Gateway-shaped backend that reads FORWARD only. <see cref="SupportsReverseRaw"/> is false; a reverse window
|
||
/// (<c>start > 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, bool boundaryAtCeilingMs = false)
|
||
: IHistorianDataSource
|
||
{
|
||
private readonly Lock _gate = new();
|
||
private int _readCount;
|
||
private int _reverseWindowCount;
|
||
private int _maxRowsReturned;
|
||
|
||
public int ReadCount => _readCount;
|
||
public int ReverseWindowCount => _reverseWindowCount;
|
||
public int MaxRowsReturned => _maxRowsReturned;
|
||
|
||
/// <summary>The earliest query start and latest query end this backend was sent.</summary>
|
||
public DateTime MinRequestedStart { get; private set; } = DateTime.MaxValue;
|
||
public DateTime MaxRequestedEnd { get; private set; } = DateTime.MinValue;
|
||
|
||
/// <summary>Reads whose row count equalled the cap (a full, possibly truncated, read).</summary>
|
||
public int FullReads { get; private set; }
|
||
|
||
/// <summary>Rows returned across every read since the last <see cref="ResetCounts"/> (#689).</summary>
|
||
public long RowsReturned { get; private set; }
|
||
|
||
/// <summary>The largest <c>maxValuesPerNode</c> asked for since the last <see cref="ResetCounts"/> (#689).</summary>
|
||
public uint MaxCapRequested { get; private set; }
|
||
|
||
public void ResetCounts()
|
||
{
|
||
lock (_gate) { FullReads = 0; RowsReturned = 0; MaxCapRequested = 0; }
|
||
Interlocked.Exchange(ref _readCount, 0);
|
||
}
|
||
|
||
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)");
|
||
}
|
||
|
||
lock (_gate)
|
||
{
|
||
if (startUtc < MinRequestedStart) MinRequestedStart = startUtc;
|
||
if (endUtc > MaxRequestedEnd) MaxRequestedEnd = endUtc;
|
||
}
|
||
|
||
static DateTime Ts(DataValueSnapshot s) => s.SourceTimestampUtc ?? DateTime.MinValue;
|
||
var prior = series.LastOrDefault(s => Ts(s) <= startUtc);
|
||
// AVEVA may stamp its fabricated row at the query start rounded UP to its millisecond resolution.
|
||
var stamp = boundaryAtCeilingMs && startUtc.Ticks % TimeSpan.TicksPerMillisecond != 0
|
||
? new DateTime(startUtc.Ticks - startUtc.Ticks % TimeSpan.TicksPerMillisecond + TimeSpan.TicksPerMillisecond, DateTimeKind.Utc)
|
||
: startUtc;
|
||
var rows = new List<DataValueSnapshot>();
|
||
if (emitBoundaryRow)
|
||
rows.Add(prior is null
|
||
? new DataValueSnapshot(0d, StatusCodes.BadNoData, stamp, stamp)
|
||
: new DataValueSnapshot(prior.Value, prior.StatusCode, stamp, stamp));
|
||
rows.AddRange(series.Where(s => Ts(s) >= startUtc && Ts(s) <= endUtc));
|
||
var page = maxValuesPerNode == 0 ? rows : rows.Take((int)maxValuesPerNode).ToList();
|
||
lock (_gate)
|
||
{
|
||
if (maxValuesPerNode > 0 && page.Count == maxValuesPerNode) FullReads++;
|
||
RowsReturned += page.Count;
|
||
if (maxValuesPerNode > MaxCapRequested) MaxCapRequested = maxValuesPerNode;
|
||
}
|
||
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
|
||
/// <see cref="TimestampResolutionTicks"/> = 10.
|
||
/// </summary>
|
||
private sealed class MicrosecondHistorianDataSource(IReadOnlyList<DataValueSnapshot> series) : IHistorianDataSource
|
||
{
|
||
private readonly SeriesHistorianDataSource _inner = new(series);
|
||
|
||
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(
|
||
string fullReference, DateTime startUtc, DateTime endUtc, uint maxValuesPerNode,
|
||
CancellationToken cancellationToken) =>
|
||
_inner.ReadRawAsync(fullReference, Truncate(startUtc), Truncate(endUtc), maxValuesPerNode, cancellationToken);
|
||
|
||
public Task<HistorianRead> ReadProcessedAsync(
|
||
string fullReference, DateTime startUtc, DateTime endUtc, TimeSpan interval,
|
||
HistoryAggregateType aggregate, CancellationToken cancellationToken) =>
|
||
_inner.ReadProcessedAsync(fullReference, startUtc, endUtc, interval, aggregate, cancellationToken);
|
||
|
||
public Task<HistorianRead> ReadAtTimeAsync(
|
||
string fullReference, IReadOnlyList<DateTime> timestampsUtc, CancellationToken cancellationToken) =>
|
||
_inner.ReadAtTimeAsync(fullReference, timestampsUtc, cancellationToken);
|
||
|
||
public Task<HistoricalEventsResult> ReadEventsAsync(
|
||
string? sourceName, DateTime startUtc, DateTime endUtc, int maxEvents,
|
||
CancellationToken cancellationToken) =>
|
||
_inner.ReadEventsAsync(sourceName, startUtc, endUtc, maxEvents, cancellationToken);
|
||
|
||
public HistorianHealthSnapshot GetHealthSnapshot() => _inner.GetHealthSnapshot();
|
||
|
||
public void Dispose() { }
|
||
}
|
||
|
||
private sealed class FixedClock(DateTime now) : TimeProvider
|
||
{
|
||
public override DateTimeOffset GetUtcNow() => new(now);
|
||
}
|
||
|
||
private async Task<(OpcUaApplicationHost Host, OtOpcUaSdkServer Server)> BootAsync()
|
||
{
|
||
var applicationUri = $"urn:OtOpcUa.HistoryPagingTest:{Guid.NewGuid():N}";
|
||
var started = await OpcUaHostStart.StartAsync(
|
||
port => new OpcUaApplicationHost(
|
||
new OpcUaApplicationHostOptions
|
||
{
|
||
ApplicationName = "OtOpcUa.HistoryPagingTest",
|
||
ApplicationUri = applicationUri,
|
||
OpcUaPort = port,
|
||
PublicHostname = "localhost",
|
||
PkiStoreRoot = _pkiRoot,
|
||
},
|
||
NullLogger<OpcUaApplicationHost>.Instance),
|
||
() => new OtOpcUaSdkServer(), Ct);
|
||
var host = started.Host;
|
||
var server = started.Server;
|
||
return (host, server);
|
||
}
|
||
|
||
public void Dispose()
|
||
{
|
||
if (Directory.Exists(_pkiRoot))
|
||
{
|
||
try { Directory.Delete(_pkiRoot, recursive: true); }
|
||
catch { /* best-effort cleanup */ }
|
||
}
|
||
}
|
||
}
|