fix(historian): #693 re-review — AtTime read cap 64 + 16 per time (bursty tags need ~6)
A time inside a gap of a bursty tag costs a 64-row jump read landing in the next burst plus a bisection back across the gap (~6 reads), so the 64 + 4/time cap refused a 1 h/day 1 Hz tag asked at noon on 30/100/300 consecutive days (186/606/1,806 reads) that AVEVA's own AtTime served. The cap is a runaway guard (HistoryReadDeadline bounds wall clock): now 64 + 16 per distinct time. Regression tests for x30/x100/x300 (each <= 7 reads a time, values checked). Docs: DateTime-hinted nodes are stepped but keep the gateway's Double (the Gateway never historizes DateTime nodes; the SQL backends return DateTime).
This commit is contained in:
+77
-2
@@ -575,6 +575,81 @@ public sealed class GatewayAtTimeTests
|
||||
ex.Message.ShouldContain("row ceiling (2");
|
||||
}
|
||||
|
||||
/// <summary>A bursty tag: 1 Hz for the first hour of every day (value = seconds since <paramref name="from"/>).</summary>
|
||||
private sealed class BurstStore(DateTime from, int days)
|
||||
{
|
||||
public long RowsStreamed;
|
||||
private const long Burst = 3600 * TimeSpan.TicksPerSecond;
|
||||
|
||||
private DateTime? FirstAtOrAfter(DateTime t)
|
||||
{
|
||||
if (t <= from) return from;
|
||||
var d = (t - from).Ticks / TimeSpan.TicksPerDay;
|
||||
var inDay = (t - from).Ticks % TimeSpan.TicksPerDay;
|
||||
var r = inDay >= Burst
|
||||
? from.AddDays(d + 1)
|
||||
: from.AddTicks(d * TimeSpan.TicksPerDay + (inDay + TimeSpan.TicksPerSecond - 1) / TimeSpan.TicksPerSecond * TimeSpan.TicksPerSecond);
|
||||
if ((r - from).Ticks % TimeSpan.TicksPerDay >= Burst) r = from.AddDays(d + 1);
|
||||
return r < from.AddDays(days) ? r : null;
|
||||
}
|
||||
|
||||
private DateTime? LastBefore(DateTime t)
|
||||
{
|
||||
if (t <= from) return null;
|
||||
var x = (t - from).Ticks - 1;
|
||||
var d = x / TimeSpan.TicksPerDay;
|
||||
var inDay = x % TimeSpan.TicksPerDay;
|
||||
var r = inDay >= Burst - TimeSpan.TicksPerSecond
|
||||
? from.AddTicks(d * TimeSpan.TicksPerDay + Burst - TimeSpan.TicksPerSecond)
|
||||
: from.AddTicks(d * TimeSpan.TicksPerDay + inDay / TimeSpan.TicksPerSecond * TimeSpan.TicksPerSecond);
|
||||
var last = from.AddDays(days - 1).AddTicks(Burst - TimeSpan.TicksPerSecond);
|
||||
return r > last ? last : r;
|
||||
}
|
||||
|
||||
private HistorianSample Row(DateTime t, uint quality = 0, DateTime? at = null) => new()
|
||||
{
|
||||
Tag = "T", Timestamp = Timestamp.FromDateTime(at ?? t), NumericValue = (t - from).Ticks / TimeSpan.TicksPerSecond,
|
||||
OpcQuality = 192, Quality = quality,
|
||||
};
|
||||
|
||||
public FakeHistorianGatewayClient Client() => new() { RawSource = Serve };
|
||||
|
||||
private IEnumerable<HistorianSample> Serve(DateTime start, DateTime end, int max)
|
||||
{
|
||||
if (max <= 0 || end < start || end > Now.AddDays(2)
|
||||
|| start.Ticks % TimeSpan.TicksPerMillisecond != 0 || end.Ticks % TimeSpan.TicksPerMillisecond != 0)
|
||||
throw new Grpc.Core.RpcException(new Grpc.Core.Status(Grpc.Core.StatusCode.InvalidArgument, "refused"));
|
||||
var rows = new List<HistorianSample>();
|
||||
if (FirstAtOrAfter(start) != start)
|
||||
rows.Add(LastBefore(start) is { } p
|
||||
? Row(p, 133, start)
|
||||
: new HistorianSample { Tag = "T", Timestamp = Timestamp.FromDateTime(start), OpcQuality = 0, Quality = 1 });
|
||||
for (var t = FirstAtOrAfter(start); t is { } u && u <= end && rows.Count < max; t = FirstAtOrAfter(u.AddTicks(1)))
|
||||
rows.Add(Row(u));
|
||||
RowsStreamed += rows.Count;
|
||||
return rows;
|
||||
}
|
||||
}
|
||||
|
||||
[Theory] // re-review: every time in a gap of a bursty tag costs a jump read + a bisection — the cap must allow that
|
||||
[InlineData(30)]
|
||||
[InlineData(100)]
|
||||
[InlineData(300)]
|
||||
public async Task A_bursty_tag_asked_at_noon_on_consecutive_days_is_served(int days)
|
||||
{
|
||||
var from = Now.Date.AddDays(-days - 1);
|
||||
var fake = new BurstStore(from, days + 1).Client();
|
||||
var times = Enumerable.Range(0, days).Select(d => from.AddDays(d).AddHours(12)).ToArray();
|
||||
var s = await AtTime(fake, times);
|
||||
for (var d = 0; d < days; d++)
|
||||
{
|
||||
double prev = d * 86400 + 3599, next = (d + 1) * 86400;
|
||||
((double)s[d].Value!).ShouldBe(prev + (next - prev) * (12 * 3600 - 3599.0) / (86400 - 3599.0), 1e-6);
|
||||
s[d].StatusCode.ShouldBe(GoodInterpolated);
|
||||
}
|
||||
fake.ReadRawCallCount.ShouldBeLessThanOrEqualTo(7 * days); // ≈ 6 a time; the cap is 64 + 16 a time
|
||||
}
|
||||
|
||||
[Fact] // L1: a small ceiling is served (paging), within the read cap
|
||||
public async Task A_small_ceiling_is_served_within_the_read_cap()
|
||||
{
|
||||
@@ -596,10 +671,10 @@ public sealed class GatewayAtTimeTests
|
||||
fake.RawStreamsCompleted.ShouldBe(3);
|
||||
}
|
||||
|
||||
[Fact] // the default cap scales with the request: 64 + 4 per distinct time
|
||||
[Fact] // the default cap scales with the request: 64 + 16 per distinct time (a runaway guard; a bursty tag needs ≈ 6)
|
||||
public async Task The_default_read_cap_is_64_plus_4_per_distinct_time()
|
||||
{
|
||||
(GatewayAtTimeReader.BaseReads, GatewayAtTimeReader.ReadsPerTime).ShouldBe((64, 4));
|
||||
(GatewayAtTimeReader.BaseReads, GatewayAtTimeReader.ReadsPerTime).ShouldBe((64, 16));
|
||||
await Task.CompletedTask;
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user