Joseph Doherty
de8bc7a12c
The #687/F4 crowding re-read doubled its room up to 8 re-reads with no cap, so a pathological single-ms cluster on a streaming gateway was streamed several times over for one page. The room is now capped at ServerHistorian:MaxTieClusterOverfetch (plumbed into GatewayHistorianDataSource as maxPreWindowRows, default 65536); past it, or once a read has already seen more, the read is BadTooManyOperations with PreWindowRows set.
370 lines
18 KiB
C#
370 lines
18 KiB
C#
using Google.Protobuf.WellKnownTypes;
|
|
using Microsoft.Extensions.Logging.Abstractions;
|
|
using Shouldly;
|
|
using Xunit;
|
|
using ZB.MOM.WW.HistorianGateway.Contracts.Grpc;
|
|
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
|
|
|
namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests;
|
|
|
|
/// <summary>
|
|
/// #687: a raw HistoryRead must return stored samples only (Part 11, <c>ReturnBounds = false</c>). An AVEVA <c>Full</c>
|
|
/// raw read starts with a synthetic row stamped exactly at the QUERY start — the value in effect there
|
|
/// (<c>Quality = 133</c>, the prior sample's value and OPC quality) or, with no earlier data, a NULL row. The fake emits
|
|
/// the REAL wire shape the live gate captured (2026-10-12): every row carries a value (histsdk puts <c>NumericValue</c>
|
|
/// on every numeric row), so a NULL row is <c>NumericValue = 0</c> + OPC quality 0 / <c>Quality = 1</c> — the evidence
|
|
/// that a row is synthetic is its timestamp (the query start), never "it has no value". A stored NULL/Bad sample looks
|
|
/// the same on the wire (<c>q=1 opc=0</c>) and IS a stored sample.
|
|
/// </summary>
|
|
public sealed class GatewayRawBoundaryRowTests
|
|
{
|
|
private static readonly CancellationToken Ct = CancellationToken.None;
|
|
|
|
/// <summary>The live gate's tag <c>HistGW.LiveTest.Gate681</c> (18:06:40 is a stored NULL).</summary>
|
|
private static readonly DateTime T0 = new(2026, 10, 12, 18, 6, 0, DateTimeKind.Utc);
|
|
|
|
private static readonly (double Sec, double Value, bool Null)[] Gate681 =
|
|
[
|
|
(0, 10, false), (5, 20, false), (20, 30, false), (40, 0, true), (50, 5, false), (60, 1, false), (90, 1, false),
|
|
(100, 7, false),
|
|
];
|
|
|
|
private static DateTime At(double seconds) => T0.AddTicks((long)(seconds * TimeSpan.TicksPerSecond));
|
|
|
|
private static HistorianSample Stored(DateTime t, double v, bool isNull) => new()
|
|
{
|
|
Tag = "T",
|
|
Timestamp = Timestamp.FromDateTime(t),
|
|
NumericValue = isNull ? 0 : v,
|
|
OpcQuality = isNull ? 0u : 192u,
|
|
Quality = isNull ? 1u : 0u,
|
|
};
|
|
|
|
/// <summary>
|
|
/// An AVEVA-shaped gateway over <paramref name="data"/>: ReadRaw over <c>[start, end]</c> ascending, capped at
|
|
/// <c>max</c>, led by the boundary row at the query start when nothing is stored exactly there.
|
|
/// </summary>
|
|
private static FakeHistorianGatewayClient Aveva(IReadOnlyList<(DateTime T, double V, bool Null)> data)
|
|
{
|
|
var sorted = data.OrderBy(d => d.T).ToList();
|
|
return new FakeHistorianGatewayClient
|
|
{
|
|
RawSource = (start, end, max) =>
|
|
{
|
|
var rows = new List<HistorianSample>();
|
|
if (sorted.All(d => d.T != start))
|
|
{
|
|
var prior = sorted.Where(d => d.T < start).Select(d => ((DateTime T, double V, bool Null)?)d).LastOrDefault();
|
|
HistorianSample boundary;
|
|
if (prior is { } p)
|
|
{
|
|
boundary = Stored(start, p.V, p.Null);
|
|
if (!p.Null) boundary.Quality = 133; // copies the prior's quality; a Bad prior stays q=1 opc=0
|
|
}
|
|
else
|
|
{
|
|
boundary = new HistorianSample { Tag = "T", Timestamp = Timestamp.FromDateTime(start), NumericValue = 0, OpcQuality = 0, Quality = 1 };
|
|
}
|
|
rows.Add(boundary);
|
|
}
|
|
rows.AddRange(sorted.Where(d => d.T >= start && d.T <= end).Select(d => Stored(d.T, d.V, d.Null)));
|
|
return max > 0 ? rows.Take(max) : rows;
|
|
},
|
|
};
|
|
}
|
|
|
|
private static FakeHistorianGatewayClient Gate() => Aveva(Gate681.Select(x => (At(x.Sec), x.Value, x.Null)).ToList());
|
|
|
|
private static GatewayHistorianDataSource Source(FakeHistorianGatewayClient fake) =>
|
|
new(fake, NullLogger<GatewayHistorianDataSource>.Instance);
|
|
|
|
[Fact]
|
|
public async Task A_window_holding_no_stored_sample_returns_nothing_not_the_value_in_effect()
|
|
{
|
|
// Live: Raw [18:06:21, 18:06:39] returned "18:06:21 v=30 Good" — a value never stored at that time.
|
|
var r = await Source(Gate()).ReadRawAsync("T", At(21), At(39), 0, Ct);
|
|
r.Samples.ShouldBeEmpty();
|
|
}
|
|
|
|
[Fact]
|
|
public async Task A_window_before_the_first_sample_returns_nothing_not_a_Bad_zero()
|
|
{
|
|
// Live: Raw [18:05:00, 18:05:30] returned "18:05:00 v=0 Bad" instead of GoodNoData.
|
|
var r = await Source(Gate()).ReadRawAsync("T", At(-60), At(-30), 10, Ct);
|
|
r.Samples.ShouldBeEmpty();
|
|
}
|
|
|
|
[Fact]
|
|
public async Task A_start_only_read_starts_at_the_first_stored_sample()
|
|
{
|
|
var r = await Source(Gate()).ReadRawAsync("T", At(30), new DateTime(DateTime.MaxValue.Ticks, DateTimeKind.Utc), 3, Ct);
|
|
r.Samples.Select(s => s.SourceTimestampUtc).ShouldBe(new DateTime?[] { At(40), At(50), At(60) });
|
|
}
|
|
|
|
[Fact]
|
|
public async Task A_stored_NULL_sample_in_the_window_is_kept_as_the_Bad_sample_it_is()
|
|
{
|
|
var r = await Source(Gate()).ReadRawAsync("T", At(30), At(55), 0, Ct);
|
|
r.Samples.Select(s => s.SourceTimestampUtc).ShouldBe(new DateTime?[] { At(40), At(50) });
|
|
r.Samples[0].StatusCode.ShouldBe(0x80000000u);
|
|
}
|
|
|
|
[Fact] // #690: the wire's value 0 on a NULL row is not a value — SqlServer/QuestDB return null + Bad for the same sample
|
|
public async Task A_stored_NULL_sample_carries_no_value_not_the_wires_zero()
|
|
{
|
|
var r = await Source(Gate()).ReadRawAsync("T", At(30), At(55), 0, Ct);
|
|
r.Samples[0].SourceTimestampUtc.ShouldBe(At(40));
|
|
r.Samples[0].Value.ShouldBeNull();
|
|
r.Samples[1].Value.ShouldBe(5d); // a Good neighbour keeps its value
|
|
}
|
|
|
|
[Fact]
|
|
public async Task A_stored_sample_exactly_at_the_window_start_is_kept()
|
|
{
|
|
var r = await Source(Gate()).ReadRawAsync("T", At(20), At(45), 0, Ct);
|
|
r.Samples.Select(s => s.SourceTimestampUtc).ShouldBe(new DateTime?[] { At(20), At(40) });
|
|
r.Samples[0].Value.ShouldBe(30d);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task The_raw_request_starts_one_millisecond_early_and_asks_for_the_boundary_allowance()
|
|
{
|
|
var fake = Gate();
|
|
await Source(fake).ReadRawAsync("T", At(30), At(55), 5, Ct);
|
|
fake.RawCalls.Single().ShouldBe((At(30) - RawAggregateReader.BoundaryMargin, At(55), 5 + RawAggregateReader.PreWindowAllowance));
|
|
}
|
|
|
|
[Fact]
|
|
public async Task A_full_page_holds_exactly_the_cap_and_the_stream_is_drained_not_abandoned()
|
|
{
|
|
var fake = Gate();
|
|
var r = await Source(fake).ReadRawAsync("T", At(0), At(200), 3, Ct);
|
|
r.Samples.Select(s => s.SourceTimestampUtc).ShouldBe(new DateTime?[] { At(0), At(5), At(20) });
|
|
fake.RawStreamsCompleted.ShouldBe(1);
|
|
}
|
|
|
|
[Theory]
|
|
[InlineData(1u)]
|
|
[InlineData(2u)]
|
|
[InlineData(3u)]
|
|
public async Task Server_style_paging_returns_every_stored_sample_once_and_nothing_else(uint cap)
|
|
{
|
|
// Ties at +60 (three samples), a stored NULL at +40, a 20-sample tie cluster 1 ms before the +60 resume point
|
|
// (crowds that resume read's margin: more than the 16-row allowance) and one 1 ms before the window start (outside
|
|
// the window; crowds the first read's margin), over the AVEVA-shaped wire.
|
|
var start = At(-10);
|
|
var end = At(200);
|
|
var inside = Gate681.Select(x => (At(x.Sec), x.Value, x.Null))
|
|
.Concat([(At(60), 2d, false), (At(60), 3d, false)])
|
|
.Concat(Enumerable.Range(0, 20).Select(i => (At(60) - TimeSpan.FromMilliseconds(1), 100d + i, false)))
|
|
.ToList();
|
|
var outside = Enumerable.Range(0, 20).Select(i => (start - TimeSpan.FromMilliseconds(1), 200d + i, false));
|
|
var data = inside;
|
|
var fake = Aveva(inside.Concat(outside).ToList());
|
|
var ds = Source(fake);
|
|
var got = new List<DataValueSnapshot>();
|
|
var skip = 0;
|
|
var cursor = start;
|
|
var resume = false;
|
|
for (var guard = 0; guard < 100; guard++)
|
|
{
|
|
// The node manager's paging (OtOpcUaNodeManager.ServeRawPagedAsync + HistoryPaging): a full page ⇒ resume at
|
|
// the last timestamp INCLUSIVE dropping the ties already returned; a short page is the last; a resume that
|
|
// cannot advance (a tie cluster larger than the cap) over-fetches the cluster with a start == end read.
|
|
var page = (await ds.ReadRawAsync("T", cursor, end, cap, Ct)).Samples;
|
|
var full = page.Count >= cap;
|
|
var samples = resume
|
|
? page.SkipWhile((x, i) => i < skip && x.SourceTimestampUtc == cursor).ToList()
|
|
: page.ToList();
|
|
if (resume && full && (samples.Count == 0 || samples[^1].SourceTimestampUtc == cursor))
|
|
{
|
|
var cluster = (await ds.ReadRawAsync("T", cursor, cursor, 1000, Ct)).Samples;
|
|
var sliceStart = Math.Min(skip, cluster.Count);
|
|
var sliceCount = Math.Min((int)cap, cluster.Count - sliceStart);
|
|
got.AddRange(cluster.Skip(sliceStart).Take(sliceCount));
|
|
if (sliceStart + sliceCount < cluster.Count)
|
|
{
|
|
skip = sliceStart + sliceCount;
|
|
continue;
|
|
}
|
|
cursor = cursor.AddTicks(1);
|
|
skip = 0;
|
|
if (cursor > end) break;
|
|
continue;
|
|
}
|
|
got.AddRange(samples);
|
|
if (!full || samples.Count == 0) break;
|
|
cursor = samples[^1].SourceTimestampUtc!.Value;
|
|
skip = samples.AsEnumerable().Reverse().TakeWhile(x => x.SourceTimestampUtc == cursor).Count();
|
|
resume = true;
|
|
}
|
|
got.Select(s => s.SourceTimestampUtc).ShouldBe(data.Select(d => (DateTime?)d.Item1).OrderBy(t => t).ToArray());
|
|
}
|
|
|
|
[Fact]
|
|
public async Task A_margin_that_swallows_the_row_budget_is_re_read_once_with_room_for_it()
|
|
{
|
|
// 30 stored rows inside the 1 ms margin (plus AVEVA's boundary row) eat the whole first budget (5 + 16): the page is
|
|
// re-read ONCE asking for cap + m + 16 (m = the pre-window rows the first read saw), which fits, and the full page
|
|
// is answered — a short page would end the client's paging early.
|
|
var margin = Enumerable.Range(0, 30).Select(i => (At(30).AddTicks(-1 - i), (double)i, false));
|
|
var window = Enumerable.Range(0, 10).Select(i => (At(31 + i), 1000d + i, false));
|
|
var fake = Aveva(margin.Concat(window).ToList());
|
|
var ds = Source(fake);
|
|
|
|
var r = await ds.ReadRawAsync("T", At(30), At(50), 5, Ct);
|
|
|
|
r.Samples.Select(s => (double)s.Value!).ShouldBe([1000d, 1001d, 1002d, 1003d, 1004d]);
|
|
fake.RawCalls.Select(c => c.MaxValues).ShouldBe([5 + 16, 5 + 21 + 16]);
|
|
fake.RawCalls.ShouldAllBe(c => c.Start == At(30) - RawAggregateReader.BoundaryMargin);
|
|
fake.RawStreamsCompleted.ShouldBe(2);
|
|
ds.GetHealthSnapshot().TotalFailures.ShouldBe(0);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task A_margin_crowding_a_truncated_read_is_re_read_with_growing_room()
|
|
{
|
|
// #694 F4: 200 rows in the margin. The first read (5 + 16) saw only 21 of them, so a single re-read sized from that
|
|
// count (5 + 21 + 16) was still short and the node failed — with no ceiling at all. The room now at least doubles per
|
|
// re-read (21, 42, 84, 168, 336), so the window is served in a bounded number of reads.
|
|
var margin = Enumerable.Range(0, 200).Select(i => (At(30).AddTicks(-1 - i), (double)i, false));
|
|
var fake = Aveva(margin.Append((At(35), 1d, false)).ToList());
|
|
var ds = Source(fake);
|
|
|
|
var r = await ds.ReadRawAsync("T", At(30), At(40), 5, Ct);
|
|
|
|
r.Samples.Select(s => s.Value).ShouldBe([1d]);
|
|
fake.ReadRawCallCount.ShouldBe(6);
|
|
fake.RawStreamsCompleted.ShouldBe(6);
|
|
ds.GetHealthSnapshot().TotalFailures.ShouldBe(0);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task A_margin_still_crowding_the_bounded_re_reads_fails_the_node_as_a_request_limit()
|
|
{
|
|
// 20,000 rows in the margin: past the bounded re-reads (21 · 2^8 ≈ 5,400 rows of room).
|
|
var margin = Enumerable.Range(0, 20_000).Select(i => (At(30).AddTicks(-1 - i), (double)i, false));
|
|
var fake = Aveva(margin.Append((At(35), 1d, false)).ToList());
|
|
var ds = Source(fake);
|
|
|
|
var ex = await Should.ThrowAsync<HistoryReadLimitExceededException>(() => ds.ReadRawAsync("T", At(30), At(40), 5, Ct));
|
|
|
|
ex.PreWindowRows.ShouldBeGreaterThan(0);
|
|
fake.ReadRawCallCount.ShouldBe(9); // the read + MaxCrowdedReReads, never more
|
|
fake.RawStreamsCompleted.ShouldBe(9);
|
|
ds.GetHealthSnapshot().TotalFailures.ShouldBe(0);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task The_crowding_re_reads_never_make_room_for_more_than_the_tie_cluster_limit()
|
|
{
|
|
// #694 R2: the room used to double up to 8 re-reads with no cap — on a streaming gateway a pathological 1M-row
|
|
// single-millisecond cluster streamed ~4M rows for one page where it used to fail at once. The room is capped at
|
|
// MaxTieClusterOverfetch (here 100); past it the read is a request limit, after a bounded number of bounded reads.
|
|
var margin = Enumerable.Range(0, 150).Select(i => (At(30).AddTicks(-1 - i), (double)i, false));
|
|
var fake = Aveva(margin.Append((At(35), 1d, false)).ToList());
|
|
var ds = new GatewayHistorianDataSource(fake, NullLogger<GatewayHistorianDataSource>.Instance, maxPreWindowRows: 100);
|
|
|
|
var ex = await Should.ThrowAsync<HistoryReadLimitExceededException>(() => ds.ReadRawAsync("T", At(30), At(40), 5, Ct));
|
|
|
|
ex.PreWindowRows.ShouldBeGreaterThan(100);
|
|
// Room 21, 42, 84: that read already saw more than 100 rows before the window, so it stops — never asking past the cap.
|
|
fake.ReadRawCallCount.ShouldBe(4);
|
|
fake.RawCalls.Max(c => c.MaxValues).ShouldBeLessThanOrEqualTo(5 + 100 + RawAggregateReader.PreWindowAllowance);
|
|
fake.RawYieldedCount.ShouldBeLessThanOrEqualTo(21 + 42 + 63 + 105);
|
|
fake.RawStreamsCompleted.ShouldBe(4);
|
|
ds.GetHealthSnapshot().TotalFailures.ShouldBe(0);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Crowding_rows_just_under_the_tie_cluster_limit_are_read_around()
|
|
{
|
|
var margin = Enumerable.Range(0, 90).Select(i => (At(30).AddTicks(-1 - i), (double)i, false));
|
|
var fake = Aveva(margin.Append((At(35), 1d, false)).ToList());
|
|
var ds = new GatewayHistorianDataSource(fake, NullLogger<GatewayHistorianDataSource>.Instance, maxPreWindowRows: 100);
|
|
|
|
var r = await ds.ReadRawAsync("T", At(30), At(40), 5, Ct);
|
|
|
|
r.Samples.Select(s => s.Value).ShouldBe([1d]);
|
|
fake.RawCalls.Max(c => c.MaxValues).ShouldBeLessThanOrEqualTo(5 + 100 + RawAggregateReader.PreWindowAllowance);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task A_short_read_is_complete_and_is_not_re_read()
|
|
{
|
|
var fake = Gate();
|
|
await Source(fake).ReadRawAsync("T", At(30), At(55), 5, Ct);
|
|
fake.ReadRawCallCount.ShouldBe(1);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task A_read_that_reached_past_the_window_end_is_complete_and_is_not_re_read()
|
|
{
|
|
// Budget 3 + 16 = 19 = boundary + 16 margin rows + the one window row + the row stored at the (exclusive) end:
|
|
// the stream is used up and the page is short, but it reached the end — every window row was seen.
|
|
var margin = Enumerable.Range(0, 16).Select(i => (At(30).AddTicks(-1 - i), (double)i, false));
|
|
var fake = Aveva(margin.Append((At(35), 1d, false)).Append((At(40), 2d, false)).ToList());
|
|
var ds = Source(fake);
|
|
|
|
var r = await ds.ReadRawAsync("T", At(30), At(40), 3, Ct);
|
|
|
|
r.Samples.Select(s => (double)s.Value!).ShouldBe([1d]);
|
|
fake.RawYieldedCount.ShouldBe(19);
|
|
fake.ReadRawCallCount.ShouldBe(1);
|
|
}
|
|
|
|
[Theory] // M1: the raw budget is NOT clamped at the gateway's default ceiling — AVEVA always emits a boundary row, so a
|
|
// page size within 16 of the ceiling would always come back short (a strict gateway refuses instead ⇒
|
|
// BadTooManyOperations, a default gateway serves it).
|
|
[InlineData(999_984u, 1_000_000)]
|
|
[InlineData(999_990u, 1_000_006)]
|
|
[InlineData(1_000_000u, 1_000_016)]
|
|
public async Task The_raw_row_budget_is_not_clamped_at_the_gateways_default_ceiling(uint cap, int expected)
|
|
{
|
|
var fake = Gate();
|
|
await Source(fake).ReadRawAsync("T", At(30), At(55), cap, Ct);
|
|
fake.RawCalls.Single().MaxValues.ShouldBe(expected);
|
|
}
|
|
|
|
// ---- L3: the forward end bound is exclusive; a point read is inclusive ----------------------------------------
|
|
|
|
[Fact]
|
|
public async Task A_stored_sample_exactly_at_the_window_end_is_not_returned()
|
|
{
|
|
var r = await Source(Gate()).ReadRawAsync("T", At(0), At(20), 0, Ct);
|
|
r.Samples.Select(s => s.SourceTimestampUtc).ShouldBe(new DateTime?[] { At(0), At(5) });
|
|
}
|
|
|
|
[Fact]
|
|
public async Task A_point_read_returns_the_samples_stored_at_that_instant()
|
|
{
|
|
var data = new List<(DateTime, double, bool)> { (At(10), 1, false), (At(20), 2, false), (At(20), 3, false), (At(30), 4, false) };
|
|
var r = await Source(Aveva(data)).ReadRawAsync("T", At(20), At(20), 1000, Ct);
|
|
r.Samples.Select(s => (double)s.Value!).ShouldBe([2d, 3d]);
|
|
}
|
|
|
|
// ---- the Unix-epoch floor ----------------------------------------------------------------------------------------
|
|
|
|
[Theory] // the gateway's time encoding starts at the Unix epoch: the 1 ms margin never reaches before it
|
|
[InlineData(0L)]
|
|
[InlineData(5_000L)]
|
|
public void The_request_start_never_precedes_the_Unix_epoch(long ticksAfterEpoch)
|
|
{
|
|
var start = DateTime.UnixEpoch.AddTicks(ticksAfterEpoch);
|
|
RawAggregateReader.RequestStartFor(start).ShouldBe(DateTime.UnixEpoch);
|
|
RawAggregateReader.RequestStartFor(At(0)).ShouldBe(At(0) - RawAggregateReader.BoundaryMargin);
|
|
}
|
|
|
|
[Theory] // with no margin left before the epoch, AVEVA's no-data row lands AT StartTime: a row at/before the epoch is
|
|
// never a stored sample (AVEVA stores nothing there — the #686 walk-back's floor too)
|
|
[InlineData(0L)]
|
|
[InlineData(-10_000_000L)]
|
|
public async Task A_read_starting_at_or_before_the_epoch_does_not_return_the_fabricated_row(long startTicksFromEpoch)
|
|
{
|
|
var epoch = DateTime.UnixEpoch;
|
|
var fake = Aveva([(epoch.AddSeconds(10), 7d, false)]);
|
|
var r = await Source(fake).ReadRawAsync("T", epoch.AddTicks(startTicksFromEpoch), epoch.AddSeconds(60), 5, Ct);
|
|
r.Samples.Select(s => s.SourceTimestampUtc).ShouldBe(new DateTime?[] { epoch.AddSeconds(10) });
|
|
}
|
|
}
|