Joseph Doherty
f966efcfad
C1 (critical): B6 computes the plan off the pinned thread while B1 made writes non-creating, so a value/status/alarm write for a node the in-flight plan adds was skipped, and change-driven drivers never republish it (BadWaitingForInitialData indefinitely; on a full rebuild every re-subscribe value landed on discarded nodes). OpcUaPublishActor now applies writes immediately AND remembers the latest per (realm, id) while a plan is in flight or queued (value+status compose; alarm state/quality coalesce to the latest, which the node's delta gate makes inert on an existing condition), replays it after each apply before answering, and drops it once the pipeline is idle. Bounded (250k keys by default; beyond it new keys are not held, logged once). I2: the data-plane-authority parse runs in its own try inside the plan task; a throw is logged and never degrades the address-space apply. I3: the SDK's DeleteMonitoredItems releases an item's component-cache reference on a Good delete; a tombstoned item's reference was already released, so the delete took one from the re-created node's entry. Released items are remembered and the orphan delete takes one back first. Event items never held a reference (EventManager + SubscribeToEvents), so the three event-item releases are gone. I4: direct tests for the HistoryUpdate override's per-request updater check and the HistoryReadEvents override's source guard (both reached by restoring the bit after a reset - the B7 race window); fixed the false remark. Minors: the MaterialiseAlarmCondition kind-swap tombstones the dropped instance; a bytes-less rebuild arriving mid-plan queues behind it (replies in order); a PlanReady/PlanFailed from a previous actor incarnation is ignored and its requester answered degraded; the tie-cluster end bound of an uncapped read fits against the backend ceiling (defence in depth - guarded by a sweep test); the lock-time benchmark stays in the default run, headroom noted. Refs #732 Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT
604 lines
30 KiB
C#
604 lines
30 KiB
C#
using Microsoft.Extensions.Logging.Abstractions;
|
||
using Opc.Ua;
|
||
using Opc.Ua.Server;
|
||
using Shouldly;
|
||
using Xunit;
|
||
using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
|
||
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>Boots ONE real SDK server for the whole bounds suite (the cases swap the historian source per read).</summary>
|
||
public sealed class BoundsServerFixture : IAsyncLifetime
|
||
{
|
||
private readonly string _pkiRoot = Path.Combine(Path.GetTempPath(), $"otopcua-bounds-{Guid.NewGuid():N}");
|
||
private OpcUaApplicationHost? _host;
|
||
|
||
/// <summary>The node manager under test.</summary>
|
||
public OtOpcUaNodeManager Nm { get; private set; } = null!;
|
||
|
||
/// <summary>The historized variable every case reads.</summary>
|
||
public NodeId NodeId { get; private set; } = null!;
|
||
|
||
/// <summary>Serialises the cases (they share the node manager's source).</summary>
|
||
public SemaphoreSlim Gate { get; } = new(1, 1);
|
||
|
||
public async ValueTask InitializeAsync()
|
||
{
|
||
var started = await OpcUaHostStart.StartAsync(
|
||
port => new OpcUaApplicationHost(
|
||
new OpcUaApplicationHostOptions
|
||
{
|
||
ApplicationName = "OtOpcUa.BoundsTest",
|
||
ApplicationUri = $"urn:OtOpcUa.BoundsTest:{Guid.NewGuid():N}",
|
||
OpcUaPort = port,
|
||
PublicHostname = "localhost",
|
||
PkiStoreRoot = _pkiRoot,
|
||
},
|
||
NullLogger<OpcUaApplicationHost>.Instance),
|
||
() => new OtOpcUaSdkServer(), CancellationToken.None);
|
||
_host = started.Host;
|
||
Nm = started.Server.NodeManager!;
|
||
Nm.HistoryContinuationStore = new InMemoryHistoryContinuationStore();
|
||
Nm.EnsureVariable("eq-1/bounds", parentFolderNodeId: null, displayName: "Bounds", dataType: "Double",
|
||
writable: false, historianTagname: "WW.Bounds", realm: AddressSpaceRealm.Uns);
|
||
NodeId = Nm.TryGetVariable("eq-1/bounds")!.NodeId;
|
||
}
|
||
|
||
public async ValueTask DisposeAsync()
|
||
{
|
||
if (_host is not null) await _host.DisposeAsync();
|
||
if (Directory.Exists(_pkiRoot))
|
||
{
|
||
try { Directory.Delete(_pkiRoot, recursive: true); }
|
||
catch { /* best-effort */ }
|
||
}
|
||
}
|
||
}
|
||
|
||
/// <summary>
|
||
/// Gitea #691: HistoryRead-Raw <c>ReturnBounds</c> (OPC UA Part 11 §4.6 bounding values, §6.5.3.2). The oracle is Part 11
|
||
/// Table 1 itself — every row, over its data set (samples at 5:00, 5:02, 5:03, 5:05, 5:06) — run against three backend
|
||
/// shapes: one that reads newest-first natively (SqlServer-like), one that reads forward only (Gateway-like, walked back
|
||
/// by <see cref="ReverseRawReader"/>) and a microsecond one (QuestDB-like,
|
||
/// truncating its window bounds).
|
||
/// </summary>
|
||
public sealed class NodeManagerHistoryReadBoundsTests(BoundsServerFixture fx) : IClassFixture<BoundsServerFixture>
|
||
{
|
||
private static readonly DateTime Day = new(2026, 1, 1, 0, 0, 0, DateTimeKind.Utc);
|
||
private static readonly DateTime Unspecified = DateTime.MinValue;
|
||
|
||
private static DateTime T(string hhmm)
|
||
{
|
||
var parts = hhmm.Split(':');
|
||
return Day.AddHours(int.Parse(parts[0])).AddMinutes(int.Parse(parts[1]));
|
||
}
|
||
|
||
/// <summary>Part 11 Table 1's data: value = minutes past 5:00.</summary>
|
||
private static DataValueSnapshot[] TableData() =>
|
||
new[] { "5:00", "5:02", "5:03", "5:05", "5:06" }
|
||
.Select(s => new DataValueSnapshot((double)(T(s) - T("5:00")).TotalMinutes, StatusCodes.Good, T(s), T(s)))
|
||
.ToArray();
|
||
|
||
public enum Shape { Reverse, ForwardOnly, Microsecond }
|
||
|
||
private static IHistorianDataSource Source(Shape shape, IReadOnlyList<DataValueSnapshot> series) => shape switch
|
||
{
|
||
Shape.Reverse => new SeriesSource(series, reverse: true, inclusiveForwardEnd: false, resolution: 1),
|
||
Shape.ForwardOnly => new SeriesSource(series, reverse: false, inclusiveForwardEnd: false, resolution: 1),
|
||
_ => new SeriesSource(series, reverse: true, inclusiveForwardEnd: false, resolution: 10),
|
||
};
|
||
|
||
/// <summary>
|
||
/// Part 11 Table 1, every row, verbatim. "FIRST"/"LAST" are <c>Bad_BoundNotFound</c> placeholders (null value) at the
|
||
/// StartTime/EndTime; a/b footnotes: one second after/before the previous returned value when that side is unspecified.
|
||
/// </summary>
|
||
public static TheoryData<string, string, uint, bool, string> Table1 => new()
|
||
{
|
||
{ "5:00", "5:05", 0, true, "5:00, 5:02, 5:03, 5:05" },
|
||
{ "5:00", "5:05", 0, false, "5:00, 5:02, 5:03" },
|
||
{ "5:01", "5:04", 0, true, "5:00, 5:02, 5:03, 5:05" },
|
||
{ "5:01", "5:04", 0, false, "5:02, 5:03" },
|
||
{ "5:05", "5:00", 0, true, "5:05, 5:03, 5:02, 5:00" },
|
||
{ "5:05", "5:00", 0, false, "5:05, 5:03, 5:02" },
|
||
{ "5:04", "5:01", 0, true, "5:05, 5:03, 5:02, 5:00" },
|
||
{ "5:04", "5:01", 0, false, "5:03, 5:02" },
|
||
{ "4:59", "5:05", 0, true, "FIRST, 5:00, 5:02, 5:03, 5:05" },
|
||
{ "4:59", "5:05", 0, false, "5:00, 5:02, 5:03" },
|
||
{ "5:01", "5:07", 0, true, "5:00, 5:02, 5:03, 5:05, 5:06, LAST" },
|
||
{ "5:01", "5:07", 0, false, "5:02, 5:03, 5:05, 5:06" },
|
||
{ "5:00", "5:05", 3, true, "5:00, 5:02, 5:03" },
|
||
{ "5:00", "5:05", 3, false, "5:00, 5:02, 5:03" },
|
||
{ "5:01", "5:04", 3, true, "5:00, 5:02, 5:03" },
|
||
{ "5:01", "5:04", 3, false, "5:02, 5:03" },
|
||
{ "5:05", "5:00", 3, true, "5:05, 5:03, 5:02" },
|
||
{ "5:05", "5:00", 3, false, "5:05, 5:03, 5:02" },
|
||
{ "5:04", "5:01", 3, true, "5:05, 5:03, 5:02" },
|
||
{ "5:04", "5:01", 3, false, "5:03, 5:02" },
|
||
{ "4:59", "5:05", 3, true, "FIRST, 5:00, 5:02" },
|
||
{ "4:59", "5:05", 3, false, "5:00, 5:02, 5:03" },
|
||
{ "5:01", "5:07", 3, true, "5:00, 5:02, 5:03" },
|
||
{ "5:01", "5:07", 3, false, "5:02, 5:03, 5:05" },
|
||
{ "5:00", "", 3, true, "5:00, 5:02, 5:03" },
|
||
{ "5:00", "", 3, false, "5:00, 5:02, 5:03" },
|
||
{ "5:00", "", 6, true, "5:00, 5:02, 5:03, 5:05, 5:06, LAST+" },
|
||
{ "5:00", "", 6, false, "5:00, 5:02, 5:03, 5:05, 5:06" },
|
||
{ "5:07", "", 6, true, "5:06, LAST+" },
|
||
{ "5:07", "", 6, false, "" },
|
||
{ "", "5:06", 3, true, "5:06, 5:05, 5:03" },
|
||
{ "", "5:06", 3, false, "5:06, 5:05, 5:03" },
|
||
{ "", "5:06", 6, true, "5:06, 5:05, 5:03, 5:02, 5:00, FIRST-" },
|
||
{ "", "5:06", 6, false, "5:06, 5:05, 5:03, 5:02, 5:00" },
|
||
{ "", "4:48", 6, true, "5:00, FIRST-" },
|
||
{ "", "4:48", 6, false, "" },
|
||
{ "4:48", "4:48", 0, true, "FIRST, 5:00" },
|
||
{ "4:48", "4:48", 0, false, "" },
|
||
{ "4:48", "4:48", 1, true, "FIRST" },
|
||
{ "4:48", "4:48", 1, false, "" },
|
||
{ "4:48", "4:48", 2, true, "FIRST, 5:00" },
|
||
{ "5:00", "5:00", 0, true, "5:00, 5:02" },
|
||
{ "5:00", "5:00", 0, false, "5:00" },
|
||
{ "5:00", "5:00", 1, true, "5:00" },
|
||
{ "5:00", "5:00", 1, false, "5:00" },
|
||
{ "5:01", "5:01", 0, true, "5:00, 5:02" },
|
||
{ "5:01", "5:01", 0, false, "" },
|
||
{ "5:01", "5:01", 1, true, "5:00" },
|
||
{ "5:01", "5:01", 1, false, "" },
|
||
};
|
||
|
||
[Theory]
|
||
[MemberData(nameof(Table1))]
|
||
public async Task Part11_table1_first_page_on_a_reverse_capable_backend(string start, string end, uint max, bool bounds, string expected) =>
|
||
await AssertFirstPage(Shape.Reverse, start, end, max, bounds, expected);
|
||
|
||
[Theory]
|
||
[MemberData(nameof(Table1))]
|
||
public async Task Part11_table1_first_page_on_a_forward_only_backend(string start, string end, uint max, bool bounds, string expected) =>
|
||
await AssertFirstPage(Shape.ForwardOnly, start, end, max, bounds, expected);
|
||
|
||
[Theory]
|
||
[MemberData(nameof(Table1))]
|
||
public async Task Part11_table1_first_page_on_a_microsecond_backend(string start, string end, uint max, bool bounds, string expected) =>
|
||
await AssertFirstPage(Shape.Microsecond, start, end, max, bounds, expected);
|
||
|
||
private async Task AssertFirstPage(Shape shape, string start, string end, uint max, bool bounds, string expected)
|
||
{
|
||
var s = start.Length == 0 ? Unspecified : T(start);
|
||
var e = end.Length == 0 ? Unspecified : T(end);
|
||
var source = Source(shape, TableData());
|
||
var (values, error, _) = await ReadAsync(source, s, e, max, bounds, cp: null);
|
||
|
||
error.StatusCode.Code.ShouldBe(StatusCodes.Good);
|
||
Render(values, s, e).ShouldBe(expected);
|
||
(source as SeriesSource)?.ReverseWindowsOnForwardOnly.ShouldBe(0);
|
||
}
|
||
|
||
/// <summary>
|
||
/// Part 11: bounds count toward NumValuesPerNode, the start bound comes with the first page and the end bound with the
|
||
/// last — so paging a bounded two-sided read at any page size concatenates to the unpaged (N = 0) answer.
|
||
/// </summary>
|
||
[Theory]
|
||
[InlineData(Shape.Reverse)]
|
||
[InlineData(Shape.ForwardOnly)]
|
||
[InlineData(Shape.Microsecond)]
|
||
public async Task Paging_a_bounded_read_concatenates_to_the_unpaged_answer(Shape shape)
|
||
{
|
||
var windows = new[]
|
||
{
|
||
("5:00", "5:05"), ("5:01", "5:04"), ("5:05", "5:00"), ("5:04", "5:01"), ("4:59", "5:05"), ("5:01", "5:07"),
|
||
("4:48", "4:48"), ("5:00", "5:00"), ("5:01", "5:01"), ("5:07", "4:00"), ("5:06", "5:06"),
|
||
};
|
||
foreach (var (start, end) in windows)
|
||
{
|
||
var source = Source(shape, TableData());
|
||
var all = await ReadAsync(source, T(start), T(end), 0, true, cp: null);
|
||
all.Cp.ShouldBeNull();
|
||
var expected = Render(all.Values, T(start), T(end));
|
||
for (uint n = 1; n <= 6; n++)
|
||
{
|
||
var paged = await DrainAsync(source, T(start), T(end), n, bounds: true);
|
||
Render(paged, T(start), T(end)).ShouldBe(expected, $"{start}->{end} paged by {n}");
|
||
}
|
||
}
|
||
}
|
||
|
||
/// <summary>One-sided reads page to the open placeholder exactly once (footnotes a/b), never past it.</summary>
|
||
[Theory]
|
||
[InlineData(Shape.Reverse)]
|
||
[InlineData(Shape.ForwardOnly)]
|
||
[InlineData(Shape.Microsecond)]
|
||
public async Task Paging_a_one_sided_bounded_read_ends_with_the_open_placeholder(Shape shape)
|
||
{
|
||
for (uint n = 1; n <= 7; n++)
|
||
{
|
||
var fwd = await DrainAsync(Source(shape, TableData()), T("5:01"), Unspecified, n, bounds: true);
|
||
Render(fwd, T("5:01"), Unspecified).ShouldBe("5:00, 5:02, 5:03, 5:05, 5:06, LAST+", $"forward by {n}");
|
||
var rev = await DrainAsync(Source(shape, TableData()), Unspecified, T("5:04"), n, bounds: true);
|
||
Render(rev, Unspecified, T("5:04")).ShouldBe("5:05, 5:03, 5:02, 5:00, FIRST-", $"reverse by {n}");
|
||
}
|
||
}
|
||
|
||
/// <summary>A bound is a stored sample of any quality: a Bad sample before StartTime is the start bound, as stored.</summary>
|
||
[Theory]
|
||
[InlineData(Shape.Reverse)]
|
||
[InlineData(Shape.ForwardOnly)]
|
||
[InlineData(Shape.Microsecond)]
|
||
public async Task A_bad_sample_is_returned_as_the_bound_as_stored(Shape shape)
|
||
{
|
||
var series = TableData();
|
||
series[0] = new DataValueSnapshot(null, StatusCodes.BadNoCommunication, T("5:00"), T("5:00"));
|
||
var (values, _, _) = await ReadAsync(Source(shape, series), T("5:01"), T("5:04"), 0, true, cp: null);
|
||
|
||
values[0].StatusCode.Code.ShouldBe(StatusCodes.BadNoCommunication);
|
||
values[0].SourceTimestamp.ShouldBe(T("5:00"));
|
||
values.Count.ShouldBe(4);
|
||
}
|
||
|
||
/// <summary>
|
||
/// Ties on a bound: the start bound before a forward window is the LAST of a tie cluster (the sample immediately
|
||
/// before StartTime in the read's order); the end bound after it is the FIRST of one; and ties ON StartTime are all
|
||
/// data (the bound is the first of them, so nothing extra is added).
|
||
/// </summary>
|
||
[Theory]
|
||
[InlineData(Shape.Reverse)]
|
||
[InlineData(Shape.ForwardOnly)]
|
||
[InlineData(Shape.Microsecond)]
|
||
public async Task Tie_clusters_on_either_side_yield_one_bound_each(Shape shape)
|
||
{
|
||
var series = new List<DataValueSnapshot>();
|
||
void Add(string t, double v) => series.Add(new DataValueSnapshot(v, StatusCodes.Good, T(t), T(t)));
|
||
Add("5:00", 1); Add("5:00", 2); Add("5:00", 3);
|
||
Add("5:02", 4);
|
||
Add("5:05", 5); Add("5:05", 6);
|
||
|
||
var (fwd, _, _) = await ReadAsync(Source(shape, series), T("5:01"), T("5:04"), 0, true, cp: null);
|
||
fwd.Select(v => (double)v.Value).ShouldBe([3d, 4d, 5d]);
|
||
|
||
var (rev, _, _) = await ReadAsync(Source(shape, series), T("5:04"), T("5:01"), 0, true, cp: null);
|
||
rev.Select(v => (double)v.Value).ShouldBe([5d, 4d, 3d]);
|
||
|
||
var (onStart, _, _) = await ReadAsync(Source(shape, series), T("5:00"), T("5:03"), 0, true, cp: null);
|
||
onStart.Select(v => (double)v.Value).ShouldBe([1d, 2d, 3d, 4d, 5d]);
|
||
}
|
||
|
||
/// <summary>
|
||
/// QuestDB truncates the window bounds it is sent to the microsecond: a sample at ⌊StartTime⌋ is inside a window that
|
||
/// starts a few ticks later, so it is data — never returned a second time as the start bound.
|
||
/// </summary>
|
||
[Fact]
|
||
public async Task A_microsecond_backend_does_not_return_a_sample_at_the_truncated_start_twice()
|
||
{
|
||
var series = TableData();
|
||
var (values, _, _) = await ReadAsync(Source(Shape.Microsecond, series), T("5:00").AddTicks(7), T("5:04"), 0, true, cp: null);
|
||
|
||
values.Select(v => (double)v.Value).ShouldBe([0d, 2d, 3d, 5d]);
|
||
}
|
||
|
||
/// <summary>
|
||
/// A backend whose forward end is inclusive (the gateway's own <c>ReadRaw</c>), or a resumed read that steps onto
|
||
/// EndTime, returns the sample AT EndTime as data — it must come back once, as the end bound.
|
||
/// </summary>
|
||
[Theory]
|
||
[InlineData(0u)]
|
||
[InlineData(1u)]
|
||
[InlineData(2u)]
|
||
[InlineData(3u)]
|
||
public async Task An_inclusive_forward_end_does_not_return_the_end_bound_twice(uint max)
|
||
{
|
||
var source = new SeriesSource(TableData(), reverse: false, inclusiveForwardEnd: true, resolution: 1);
|
||
var values = await DrainAsync(source, T("5:00"), T("5:05"), max, bounds: true);
|
||
Render(values, T("5:00"), T("5:05")).ShouldBe("5:00, 5:02, 5:03, 5:05");
|
||
}
|
||
|
||
/// <summary>
|
||
/// #736 (H1 fuzz) — a backend row ceiling sizes the page BELOW NumValuesPerNode (RawPageLimit: ceiling 4 ⇒ 3 data
|
||
/// rows per page). When the data exactly fills that page, the end bound must go to the NEXT page through the
|
||
/// continuation point, not be appended to a page that is already at its effective cap: the fit used to be
|
||
/// judged against NumValuesPerNode (10 here), so the page came back with cap + 1 values.
|
||
/// </summary>
|
||
[Fact]
|
||
public async Task A_full_ceiling_sized_page_carries_the_end_bound_on_the_next_page()
|
||
{
|
||
var source = new SeriesSource(TableData(), reverse: true, inclusiveForwardEnd: false, resolution: 1, ceiling: 4);
|
||
var (first, error, cp) = await ReadAsync(source, T("5:01"), T("5:06"), max: 10, bounds: true, cp: null);
|
||
error.StatusCode.Code.ShouldBe(StatusCodes.Good);
|
||
|
||
// Start bound 5:00 + the three data rows the ceiling allows = the page's effective cap, exactly.
|
||
Render(first, T("5:01"), T("5:06")).ShouldBe("5:00, 5:02, 5:03, 5:05");
|
||
cp.ShouldNotBeNull("the page is full, so the end bound is owed on the next page");
|
||
|
||
var (second, error2, cp2) = await ReadAsync(source, T("5:01"), T("5:06"), max: 10, bounds: true, cp);
|
||
error2.StatusCode.Code.ShouldBe(StatusCodes.Good);
|
||
Render(second, T("5:01"), T("5:06")).ShouldBe("5:06");
|
||
cp2.ShouldBeNull();
|
||
|
||
// Control: without a ceiling the same read fits on ONE page of NumValuesPerNode — the fit keys on the
|
||
// effective cap, not on a smaller constant.
|
||
var (whole, _, cpWhole) = await ReadAsync(Source(Shape.Reverse, TableData()), T("5:01"), T("5:06"), max: 10, bounds: true, cp: null);
|
||
Render(whole, T("5:01"), T("5:06")).ShouldBe("5:00, 5:02, 5:03, 5:05, 5:06");
|
||
cpWhole.ShouldBeNull();
|
||
}
|
||
|
||
/// <summary>
|
||
/// Lane B review fix round 1 (minor): an UNCAPPED read (N = 0) against a backend with a row ceiling L pages at
|
||
/// L, and MaxReturnDataValues advertises L — so no page, the tie-cluster slices and their end bound included,
|
||
/// may carry more than L values; and the pages still concatenate to the uncapped single-shot answer. Swept
|
||
/// over ceilings and tie-cluster sizes up to what one over-fetch read can hold (L − 1).
|
||
/// </summary>
|
||
[Fact]
|
||
public async Task An_uncapped_read_never_returns_a_page_larger_than_the_backend_ceiling()
|
||
{
|
||
foreach (var ceiling in new[] { 2, 3, 4, 6 })
|
||
{
|
||
for (var ties = 1; ties < ceiling; ties++)
|
||
{
|
||
var series = new List<DataValueSnapshot> { Snap("5:00", 0) };
|
||
for (var i = 0; i < ties; i++) series.Add(Snap("5:02", 10 + i));
|
||
series.Add(Snap("5:03", 3));
|
||
for (var i = 0; i < ties; i++) series.Add(Snap("5:05", 50 + i));
|
||
series.Add(Snap("5:06", 6));
|
||
|
||
foreach (var (start, end) in new[] { ("5:01", "5:04"), ("5:00", "5:06"), ("5:01", "5:07"), ("5:06", "5:01") })
|
||
{
|
||
var expected = Render(
|
||
(await ReadAsync(new SeriesSource(series, reverse: true, inclusiveForwardEnd: false, resolution: 1), T(start), T(end), 0, true, null)).Values,
|
||
T(start), T(end));
|
||
|
||
var capped = new SeriesSource(series, reverse: true, inclusiveForwardEnd: false, resolution: 1, ceiling: ceiling);
|
||
var all = new List<DataValue>();
|
||
byte[]? cp = null;
|
||
var pages = 0;
|
||
do
|
||
{
|
||
var (values, error, next) = await ReadAsync(capped, T(start), T(end), 0, true, cp);
|
||
error.StatusCode.Code.ShouldBe(StatusCodes.Good, $"L={ceiling} ties={ties} {start}->{end}");
|
||
values.Count.ShouldBeLessThanOrEqualTo(ceiling, $"L={ceiling} ties={ties} {start}->{end} page {pages}");
|
||
all.AddRange(values);
|
||
cp = next;
|
||
(++pages).ShouldBeLessThan(60, "paging must terminate");
|
||
}
|
||
while (cp is not null);
|
||
|
||
Render(all, T(start), T(end)).ShouldBe(expected, $"L={ceiling} ties={ties} {start}->{end}");
|
||
}
|
||
}
|
||
}
|
||
|
||
static DataValueSnapshot Snap(string at, double v) => new(v, StatusCodes.Good, T(at), T(at));
|
||
}
|
||
|
||
/// <summary>
|
||
/// #691 review: no data AND no bound found ⇒ <c>GoodNoData</c> with no values — the answer a tag with no history (and a
|
||
/// disabled historian) gave before bounds (§6.5.3.2: bounds make the result Good only "if one or both exist"). Only
|
||
/// when the WHOLE read is empty: a single bound found keeps the placeholders (Table 1: "4:48 4:48 1 ⇒ FIRST").
|
||
/// </summary>
|
||
[Theory]
|
||
[InlineData(Shape.Reverse)]
|
||
[InlineData(Shape.ForwardOnly)]
|
||
[InlineData(Shape.Microsecond)]
|
||
public async Task No_data_and_no_bound_found_is_GoodNoData_with_no_values(Shape shape)
|
||
{
|
||
var windows = new (string S, string E)[] { ("5:00", "5:05"), ("5:05", "5:00"), ("5:00", "5:00"), ("5:00", ""), ("", "5:05") };
|
||
foreach (var (start, end) in windows)
|
||
{
|
||
var s = start.Length == 0 ? Unspecified : T(start);
|
||
var e = end.Length == 0 ? Unspecified : T(end);
|
||
foreach (var n in new uint[] { 0, 1, 2, 3 })
|
||
{
|
||
if (n == 0 && (s == Unspecified || e == Unspecified)) continue; // a one-sided read needs a cap
|
||
var (values, error, cp) = await ReadResultAsync(Source(shape, []), s, e, n);
|
||
values.ShouldBeEmpty($"{start}->{end} by {n}");
|
||
cp.ShouldBeNull();
|
||
error.StatusCode.Code.ShouldBe(StatusCodes.Good);
|
||
LastNodeStatus.Code.ShouldBe(StatusCodes.GoodNoData, $"{start}->{end} by {n}");
|
||
}
|
||
}
|
||
|
||
// Positive control: one bound found keeps the missing one's placeholder.
|
||
var single = await DrainAsync(Source(shape, TableData()), T("4:48"), T("4:48"), 1, bounds: true);
|
||
Render(single, T("4:48"), T("4:48")).ShouldBe("FIRST, 5:00");
|
||
}
|
||
|
||
/// <summary>
|
||
/// #691 review: <c>ServerHistorian:HonorReturnBounds = false</c> restores the pre-#691 answer — ReturnBounds is ignored
|
||
/// (Table 1's "No" rows for every "Yes" row) and a modified read with ReturnBounds is not refused for it.
|
||
/// </summary>
|
||
[Fact]
|
||
public async Task The_kill_switch_ignores_return_bounds()
|
||
{
|
||
await fx.Gate.WaitAsync(TestContext.Current.CancellationToken);
|
||
try
|
||
{
|
||
fx.Nm.HonorReturnBounds = false;
|
||
fx.Nm.HistorianDataSource = Source(Shape.Reverse, TableData());
|
||
var (values, _, _) = Read(T("5:01"), T("5:07"), 0, bounds: true, cp: null, modified: false);
|
||
Render(values, T("5:01"), T("5:07")).ShouldBe("5:02, 5:03, 5:05, 5:06");
|
||
|
||
var (_, modErr, _) = Read(T("5:00"), T("5:05"), 0, bounds: true, cp: null, modified: true);
|
||
modErr.StatusCode.Code.ShouldBe(StatusCodes.BadHistoryOperationUnsupported, "no trail on this fake — not InvalidArgument");
|
||
}
|
||
finally
|
||
{
|
||
fx.Nm.HonorReturnBounds = true;
|
||
fx.Gate.Release();
|
||
}
|
||
}
|
||
|
||
[Fact]
|
||
public async Task Bounds_are_honoured_by_default()
|
||
{
|
||
await fx.Gate.WaitAsync(TestContext.Current.CancellationToken);
|
||
try { fx.Nm.HonorReturnBounds.ShouldBeTrue(); }
|
||
finally { fx.Gate.Release(); }
|
||
}
|
||
|
||
/// <summary>Part 11 §6.5.3.3: a modified read with ReturnBounds is <c>Bad_InvalidArgument</c>.</summary>
|
||
[Fact]
|
||
public async Task A_modified_read_with_return_bounds_is_an_invalid_argument()
|
||
{
|
||
await fx.Gate.WaitAsync(TestContext.Current.CancellationToken);
|
||
try
|
||
{
|
||
fx.Nm.HistorianDataSource = Source(Shape.Reverse, TableData());
|
||
var (_, error, cp) = Read(T("5:00"), T("5:05"), 0, bounds: true, cp: null, modified: true);
|
||
error.StatusCode.Code.ShouldBe(StatusCodes.BadInvalidArgument);
|
||
cp.ShouldBeNull();
|
||
}
|
||
finally { fx.Gate.Release(); }
|
||
}
|
||
|
||
// --- helpers --------------------------------------------------------------------------------
|
||
|
||
private static string Render(IReadOnlyList<DataValue> values, DateTime start, DateTime end)
|
||
{
|
||
var parts = new List<string>();
|
||
for (var i = 0; i < values.Count; i++)
|
||
{
|
||
var v = values[i];
|
||
if (v.StatusCode.Code != StatusCodes.BadBoundNotFound)
|
||
{
|
||
v.Value.ShouldNotBeNull();
|
||
parts.Add($"{v.SourceTimestamp.Hour}:{v.SourceTimestamp.Minute:00}");
|
||
continue;
|
||
}
|
||
|
||
v.Value.ShouldBeNull("a missing bound carries a null value");
|
||
var isFirst = i == 0 && start != Unspecified && v.SourceTimestamp == start;
|
||
var isLast = end != Unspecified && v.SourceTimestamp == end;
|
||
if (isFirst) { parts.Add("FIRST"); continue; }
|
||
if (isLast) { parts.Add("LAST"); continue; }
|
||
// Open side (footnotes a/b): one second from the previous returned value.
|
||
i.ShouldBeGreaterThan(0);
|
||
var prev = values[i - 1].SourceTimestamp;
|
||
if (v.SourceTimestamp == prev.AddSeconds(1)) parts.Add("LAST+");
|
||
else if (v.SourceTimestamp == prev.AddSeconds(-1)) parts.Add("FIRST-");
|
||
else parts.Add($"?{v.SourceTimestamp:O}");
|
||
}
|
||
|
||
return string.Join(", ", parts);
|
||
}
|
||
|
||
private async Task<List<DataValue>> DrainAsync(IHistorianDataSource source, DateTime start, DateTime end, uint max, bool bounds)
|
||
{
|
||
var all = new List<DataValue>();
|
||
byte[]? cp = null;
|
||
var pages = 0;
|
||
do
|
||
{
|
||
var (values, error, next) = await ReadAsync(source, start, end, max, bounds, cp);
|
||
error.StatusCode.Code.ShouldBe(StatusCodes.Good);
|
||
if (max > 0) values.Count.ShouldBeLessThanOrEqualTo((int)max, "bounds count toward NumValuesPerNode");
|
||
all.AddRange(values);
|
||
cp = next;
|
||
(++pages).ShouldBeLessThan(50, "paging must terminate");
|
||
}
|
||
while (cp is not null);
|
||
|
||
return all;
|
||
}
|
||
|
||
/// <summary>The node-level StatusCode of the last <see cref="Read"/> (GoodNoData vs Good).</summary>
|
||
private StatusCode LastNodeStatus;
|
||
|
||
private async Task<(IReadOnlyList<DataValue> Values, ServiceResult Error, byte[]? Cp)> ReadResultAsync(
|
||
IHistorianDataSource source, DateTime start, DateTime end, uint max) =>
|
||
await ReadAsync(source, start, end, max, bounds: true, cp: null);
|
||
|
||
private async Task<(IReadOnlyList<DataValue> Values, ServiceResult Error, byte[]? Cp)> ReadAsync(
|
||
IHistorianDataSource source, DateTime start, DateTime end, uint max, bool bounds, byte[]? cp)
|
||
{
|
||
await fx.Gate.WaitAsync(TestContext.Current.CancellationToken);
|
||
try
|
||
{
|
||
fx.Nm.HistorianDataSource = source;
|
||
return Read(start, end, max, bounds, cp, modified: false);
|
||
}
|
||
finally { fx.Gate.Release(); }
|
||
}
|
||
|
||
private (IReadOnlyList<DataValue> Values, ServiceResult Error, byte[]? Cp) Read(
|
||
DateTime start, DateTime end, uint max, bool bounds, byte[]? cp, bool modified)
|
||
{
|
||
var context = new OperationContext(new RequestHeader(), secureChannelContext: null, RequestType.HistoryRead, identity: null);
|
||
var details = new ReadRawModifiedDetails
|
||
{
|
||
StartTime = start, EndTime = end, NumValuesPerNode = max, ReturnBounds = bounds, IsReadModified = modified,
|
||
};
|
||
var nodesToRead = new List<HistoryReadValueId> { new() { NodeId = fx.NodeId, ContinuationPoint = cp } };
|
||
var results = new List<SdkHistoryReadResult> { null! };
|
||
var errors = new List<ServiceResult> { null! };
|
||
fx.Nm.HistoryRead(context, details, TimestampsToReturn.Both, releaseContinuationPoints: false, nodesToRead, results, errors);
|
||
LastNodeStatus = results[0]?.StatusCode ?? StatusCodes.Bad;
|
||
|
||
var values = new List<DataValue>();
|
||
if (results[0]?.HistoryData is { } ho && !ExtensionObject.IsNull(ho))
|
||
values.AddRange(((HistoryData)ExtensionObject.ToEncodeable(ho)).DataValues);
|
||
return (values, errors[0], results[0]?.ContinuationPoint);
|
||
}
|
||
|
||
/// <summary>
|
||
/// A series-backed historian. <paramref name="reverse"/>: serves <c>start > end</c> as <c>(end, start]</c>
|
||
/// newest-first (SqlServer/QuestDB); otherwise rejects it (the Gateway) and counts that. The forward read is
|
||
/// <c>[start, end)</c>, or <c>[start, end]</c> when <paramref name="inclusiveForwardEnd"/> (the gateway's own end);
|
||
/// a point read is the instant. <paramref name="resolution"/> > 1 truncates every window bound to it first (QuestDB
|
||
/// µs); the series must already be stored at that resolution.
|
||
/// </summary>
|
||
private sealed class SeriesSource(IReadOnlyList<DataValueSnapshot> series, bool reverse, bool inclusiveForwardEnd, long resolution, int ceiling = 0)
|
||
: IHistorianDataSource
|
||
{
|
||
public int ReverseWindowsOnForwardOnly;
|
||
|
||
public bool SupportsReverseRaw => reverse;
|
||
|
||
public long TimestampResolutionTicks => resolution;
|
||
|
||
/// <summary>A strict row ceiling (0 = none), so a page is sized below NumValuesPerNode (RawPageLimit).</summary>
|
||
public int MaxRawValuesPerRead => ceiling;
|
||
|
||
private DateTime Trunc(DateTime t) => new(t.Ticks - t.Ticks % resolution, DateTimeKind.Utc);
|
||
|
||
public Task<HistorianRead> ReadRawAsync(
|
||
string fullReference, DateTime startUtc, DateTime endUtc, uint maxValuesPerNode, CancellationToken cancellationToken)
|
||
{
|
||
static DateTime Ts(DataValueSnapshot s) => s.SourceTimestampUtc ?? DateTime.MinValue;
|
||
startUtc = Trunc(startUtc);
|
||
endUtc = Trunc(endUtc);
|
||
List<DataValueSnapshot> window;
|
||
if (startUtc > endUtc)
|
||
{
|
||
if (!reverse)
|
||
{
|
||
Interlocked.Increment(ref ReverseWindowsOnForwardOnly);
|
||
throw new InvalidOperationException("start_time must be less than or equal to end_time. (InvalidArgument)");
|
||
}
|
||
|
||
window = series.Where(s => Ts(s) > endUtc && Ts(s) <= startUtc).Reverse().ToList();
|
||
}
|
||
else if (startUtc == endUtc) window = series.Where(s => Ts(s) == startUtc).ToList();
|
||
else window = series.Where(s => Ts(s) >= startUtc && (inclusiveForwardEnd ? Ts(s) <= endUtc : Ts(s) < endUtc)).ToList();
|
||
|
||
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() { }
|
||
}
|
||
}
|