Files
lmxopcua/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/ReverseWalk.cs
T
Joseph Doherty 5897764024
v2-ci / build (pull_request) Successful in 5m6s
v2-ci / unit-tests (pull_request) Failing after 29m54s
fix(historian): fail fast for missing tags and reduce gateway read cost
2026-10-13 04:31:52 -04:00

455 lines
24 KiB
C#
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
namespace ZB.MOM.WW.OtOpcUa.OpcUaServer;
/// <summary>What one walked-back page tells the next page of the same read (carried in the continuation point).</summary>
/// <param name="WidthTicks">The segment width the walk ended on, aimed at the page's density — the next page's first width (0 = none).</param>
/// <param name="EmptyFromTicks">
/// With <paramref name="EmptyToTicks"/>: a stretch <c>(EmptyFrom, EmptyTo)</c> (both exclusive) a complete segment read
/// proved holds no row (#691) — typically from the oldest row the page saw down to the walk's floor. A resumed page
/// steps straight over it instead of re-walking it. Both 0 = none.
/// </param>
/// <param name="EmptyToTicks">The exclusive upper end of the empty stretch.</param>
internal readonly record struct ReverseWalkHint(long WidthTicks, long EmptyFromTicks = 0, long EmptyToTicks = 0)
{
/// <summary>Whether an empty stretch is carried.</summary>
public bool HasEmptyStretch => EmptyToTicks > EmptyFromTicks && EmptyToTicks > 0;
}
/// <summary>One forward read of a segment: the rows (ascending by time) and whether a cap cut the read short.</summary>
/// <typeparam name="T">Row type.</typeparam>
/// <param name="Rows">Rows in ascending time order (ties in any order).</param>
/// <param name="Truncated">A cap stopped the read — more rows may follow the last one.</param>
internal readonly record struct ForwardSegmentRead<T>(IReadOnlyList<T> Rows, bool Truncated);
/// <summary>
/// Gitea #686/#689: the walk-back that serves a Part 11 reverse read — <c>(lo, hi]</c> newest-first, capped — from a
/// backend that can only read FORWARD. Shared by raw values (<see cref="ReverseRawReader"/>) and events; the row type
/// only has to expose a timestamp.
/// </summary>
/// <remarks>
/// <para>
/// <b>Uncapped</b> (<c>want = 0</c>): ONE forward read of the whole window, filtered and reversed in memory.
/// </para>
/// <para>
/// <b>Capped: walk back in segments.</b> Starting at <c>hi</c>, read one segment <c>(segLo, segHi]</c> FORWARD, keep
/// it newest-first, and step to the next older segment until the cap is met or <c>lo</c> is reached. The first
/// segment is <see cref="InitialSegment"/> wide, or the width the previous page of the same read ended on (carried in
/// the continuation point). After each read the next width is re-aimed from the density just observed at about half
/// the PAGE budget <c>2N + 16</c> (≈ N + 8 rows), growing at most ×4 per step (#689 — it used to aim at half the tie
/// budget, ~32k rows, whatever N was) — so a live tag's "latest N" is usually one read of about N rows, and sparse
/// history is found in a few dozen tiny reads.
/// </para>
/// <para>
/// <b>Read sizes.</b> A segment read asks for its budget + 2 rows (room for a row an AVEVA read fabricates at its query
/// start, and one to see past the segment). The budget is the page budget, except a <b>discovery</b> segment — the
/// first of a page with no carried width, or one after an empty segment (whose ×4 growth may overshoot into dense
/// data) — which may read up to <see cref="DiscoveryRows"/>. The tie budget <c>M = max(1, MaxTieClusterOverfetch)</c>
/// is only a CEILING: a single timestamp holding more rows than the segment budget escalates to <c>M</c>.
/// </para>
/// <para>
/// <b>Never answer from a truncated read.</b> Rows stream ascending, so a read is complete for its segment when it was
/// not cut short OR its last row is already past <c>segHi</c>. An incomplete read is discarded: when rows BEFORE the
/// segment (a neighbouring tie cluster inside the alignment slack, a fabricated row) crowded it out it is re-read once
/// with room for them (at most twice the budget + 4); otherwise the segment is narrowed — to the density the truncated
/// read showed, and never again below the newest row it saw (<c>known</c>: rows exist there, so walking past it before
/// finding what lies above it would re-read the same dense stretch over and over; the gap above it is bisected
/// instead). Only more than <c>M</c> rows within one millisecond — a segment no wider than the 1 ms request
/// granularity that still does not fit at the tie budget — is refused (per millisecond, not per timestamp: narrower
/// requests are the same request, and the Gateway stores milliseconds).
/// </para>
/// <para>
/// <b>Proving the rest empty (#691).</b> Walking down to the floor through empty history used to take one read per ×4
/// growth step (~14 reads from a minute to the epoch) — on every page that ran out of rows. After two consecutive empty
/// segments the walk asks once per page for the WHOLE remainder <c>(lo, segHi]</c> at the page budget: a complete
/// answer ends the walk in one read; a truncated one (deep history) is discarded and the walk carries on as before. And
/// the stretch a page's last complete segment proved empty below its oldest row rides the continuation point
/// (<see cref="ReverseWalkHint.EmptyFromTicks"/>), so the next page steps over it — the rows of a page are often the
/// last ones before a stretch the previous page already read.
/// </para>
/// <para>
/// <b>Edges.</b> A segment is requested as <c>[⌊segLo⌋, ⌈segHi + 1 tick⌉]</c>, both rounded outward to the millisecond
/// (the gateway's / AVEVA's resolution), and filtered to <c>(segLo, segHi]</c> on the rows' own timestamps — exact
/// whether the backend's forward end is exclusive or inclusive and whether it truncates its bounds; a row fabricated at
/// the (millisecond) query start can never land inside the segment (#687). Adjacent segments share no timestamp.
/// </para>
/// <para>
/// <b>Ties</b> come back in the backend's forward order reversed, or — with a tie order given (events, whose gateway
/// order within a timestamp is arbitrary) — sorted by it and reversed, so a resumed page's boundary-tie skip lines up.
/// </para>
/// <para>
/// <b>Bounds.</b> The walk never asks for data after now + <see cref="LookAhead"/> nor before <see cref="Floor"/> (the
/// Unix epoch — AVEVA holds nothing earlier). A walked-back read returns only rows strictly after the epoch and no
/// later than now + one day.
/// </para>
/// </remarks>
internal static class ReverseWalk
{
/// <summary>The first segment's width when no hint is carried.</summary>
internal static readonly TimeSpan InitialSegment = TimeSpan.FromMinutes(1);
/// <summary>The walk's exclusive lower limit.</summary>
internal static readonly DateTime Floor = DateTime.UnixEpoch;
/// <summary>How far past "now" the walk starts at most.</summary>
internal static readonly TimeSpan LookAhead = TimeSpan.FromDays(1);
/// <summary>Slack added to a page's row budget (#689): room for density variation between segments.</summary>
internal const int PageSlack = 16;
/// <summary>The most rows a discovery segment (see the remarks) may read before it counts as truncated (#689).</summary>
internal const int DiscoveryRows = 1024;
/// <summary>Request bounds are rounded outward to this (the gateway / AVEVA timestamp resolution).</summary>
private const long Alignment = TimeSpan.TicksPerMillisecond;
/// <summary>The rows a segment is sized for: <c>2N + </c><see cref="PageSlack"/>, never above the tie budget.</summary>
/// <param name="want">The page cap N (&gt; 0).</param>
/// <param name="tieBudget">The tie budget M (&gt;= 1).</param>
/// <returns>The page budget.</returns>
internal static int PageBudget(uint want, int tieBudget) => (int)Math.Clamp(2L * want + PageSlack, 1L, Math.Max(1, tieBudget));
/// <summary>Reads <c>(lo, hi]</c> newest-first from forward reads.</summary>
/// <typeparam name="T">Row type.</typeparam>
/// <param name="read">One forward read <c>[start, end]</c> asking for at most <c>cap</c> rows (<c>0</c> = all).</param>
/// <param name="ticksOf">A row's timestamp, in ticks.</param>
/// <param name="tieOrder">Order within one timestamp (null = the backend's forward order).</param>
/// <param name="hi">The later (inclusive) bound.</param>
/// <param name="lo">The earlier (exclusive) bound.</param>
/// <param name="want">Cap; <c>0</c> = every row in the window (one forward read).</param>
/// <param name="tieBudget">The tie budget M (values below 1 count as 1).</param>
/// <param name="resolutionTicks">The backend's timestamp resolution (&gt;= 1).</param>
/// <param name="nowUtc">Now, for the look-ahead clamp.</param>
/// <param name="hint">What the previous page of the same read carried (default = none).</param>
/// <param name="tooManyAtOneTimestamp">Builds the exception for a timestamp holding more than M rows.</param>
/// <param name="ct">Cancellation.</param>
/// <param name="useDiscoveryBudgetForGaps">Use ceiling-sized discovery and retain truncated remainder rows (#696, strict raw backends only).</param>
/// <returns>At most <paramref name="want"/> rows, newest first, and the hint for the next page.</returns>
public static async Task<(List<T> Rows, ReverseWalkHint Next)> WalkAsync<T>(
Func<DateTime, DateTime, int, CancellationToken, Task<ForwardSegmentRead<T>>> read,
Func<T, long> ticksOf,
IComparer<T>? tieOrder,
DateTime hi, DateTime lo, uint want, int tieBudget, long resolutionTicks, DateTime nowUtc,
ReverseWalkHint hint, Func<DateTime, Exception> tooManyAtOneTimestamp, CancellationToken ct, bool useDiscoveryBudgetForGaps = false)
{
var resolution = Math.Max(1L, resolutionTicks);
var granularity = Math.Max(Alignment, resolution);
var budget = Math.Max(1, tieBudget);
var result = new List<T>();
var horizon = nowUtc.Ticks > DateTime.MaxValue.Ticks - LookAhead.Ticks ? DateTime.MaxValue.Ticks : nowUtc.Ticks + LookAhead.Ticks;
var hiTicks = Math.Min(hi.Ticks, horizon);
var stopTicks = Math.Max(lo.Ticks, Floor.Ticks);
if (hiTicks <= stopTicks) return (result, hint);
if (want == 0)
{
// One forward read of the whole window, reversed — the uncapped forward read's own cost and bound.
var all = (await read(Utc(AlignDown(stopTicks, granularity)), Utc(AlignUp(hiTicks + 1, granularity)), 0, ct)
.ConfigureAwait(false)).Rows;
AppendNewestFirst(all, ticksOf, tieOrder, stopTicks, hiTicks, result, int.MaxValue);
return (result, hint);
}
var maxWidth = DateTime.MaxValue.Ticks;
var discovering = hint.WidthTicks <= 0;
var width = discovering ? Math.Max(resolution, InitialSegment.Ticks) : Math.Clamp(hint.WidthTicks, resolution, maxWidth);
var pageBudget = PageBudget(want, budget);
var discoveryBudget = Math.Min(budget, Math.Max(pageBudget, DiscoveryRows));
// `known` > 0: a truncated read saw rows up to this timestamp (below segHi), and nothing is known of (known, segHi].
// Segments stay above it until that stretch is read; then `knownWidth` (the density the truncated read showed)
// sizes the segment that ends on it. Searching the gap: first a top-anchored segment at that density (data running
// right up to segHi); if it is empty, the data stops somewhere above `known`, so GALLOP up from it — read
// (known + lowerBy, segHi] with lowerBy growing ×4, `known` following what each truncated read saw — until a read
// comes back empty (the top is now bracketed below its segLo), then BISECT that bracket. Any complete read holding
// rows ends the search. A burst that ended long before segHi is found in a few reads instead of bisecting the whole
// gap down to the millisecond, and a long dense stretch is crossed geometrically, not a budget at a time (#691.4
// review). `noGallop` falls back to plain top-anchored halving if a galloping read is crowded by rows before it.
var known = 0L;
var knownWidth = 0L;
var lowerBy = 0L;
var bisecting = false;
var noGallop = false;
var segHi = hiTicks;
// #691: the remainder probe (once per page, after two empty segments) and the empty stretches.
var emptyRun = 0;
var probed = false;
var carried = hint.HasEmptyStretch ? (hint.EmptyFromTicks, hint.EmptyToTicks) : (0L, 0L);
(long From, long To) proven = (0L, 0L);
while (segHi > stopTicks && result.Count < want)
{
if (carried.Item2 > carried.Item1 && segHi > carried.Item1 && segHi < carried.Item2)
{
segHi = carried.Item1; // (from, segHi] lies inside a stretch a previous page proved empty
continue;
}
var probe = !probed && known == 0 && emptyRun >= 2 && segHi - stopTicks > width;
var w = probe ? segHi - stopTicks : width;
var lowered = false;
var bisected = false;
if (known > 0)
{
var gap = segHi - known;
lowered = lowerBy > 0 && lowerBy < gap;
bisected = !lowered && (bisecting || lowerBy > 0) && gap > 2 * Math.Max(granularity, knownWidth);
// Never narrower than the request granularity: requests are rounded outward to it, so a narrower segment
// re-sends the same request — and a width left at one tick (a tie cluster's) stepped segHi down a tick per
// read, ~10,000 identical reads (#691.4 re-review). Keep the Min: a truncated read crowded by rows before
// the segment must still shrink the segment.
w = lowered ? gap - lowerBy
: bisected ? gap / 2
: Math.Min(Math.Max(w, granularity), gap <= 2 * granularity ? gap : gap / 2);
}
var segLo = segHi - stopTicks <= w ? stopTicks : segHi - w;
var start = Utc(AlignDown(segLo, granularity));
var end = Utc(AlignUp(segHi + 1, granularity));
var segBudget = (discovering && !probe) || (useDiscoveryBudgetForGaps && (probe || known > 0))
? discoveryBudget : pageBudget;
var cap = segBudget + 2;
IReadOnlyList<T> rows;
var narrowed = false;
while (true)
{
ct.ThrowIfCancellationRequested();
var segment = await read(start, end, cap, ct).ConfigureAwait(false);
rows = segment.Rows;
if (!segment.Truncated || rows.Count == 0 || ticksOf(rows[^1]) > segHi) break; // complete for (segLo, segHi]
if (probe)
{
// #696: strict-ceiling discovery retains the rows/density instead of discarding the remainder probe.
probed = true;
if (!useDiscoveryBudgetForGaps)
{
narrowed = true;
break;
}
probe = false;
}
var leading = 0;
var inner = 0;
var firstInner = 0L;
var lastInner = 0L;
foreach (var r in rows)
{
var ts = ticksOf(r);
if (ts <= segLo) leading++;
else if (ts <= segHi)
{
if (inner++ == 0) firstInner = ts;
lastInner = ts;
}
}
// Crowded out by rows before the segment, with the segment itself possibly within budget: re-read once
// with room for them.
var roomier = leading + segBudget + 2;
if (leading > 0 && inner <= segBudget && roomier > cap && roomier <= 2 * segBudget + 4)
{
cap = roomier;
continue;
}
if (segHi - segLo <= granularity)
{
// A segment no wider than the request granularity (1 ms — narrower requests are the same request) still
// over its budget: a tie cluster, or a burst inside one millisecond. Escalate to the tie budget M before
// refusing — up to M rows within one millisecond are served (#686); more are refused, per millisecond
// (#691.4 re-review — exact for the millisecond-resolution Gateway).
if (segBudget < budget)
{
segBudget = budget;
cap = segBudget + 2;
continue;
}
throw tooManyAtOneTimestamp(Utc(segHi));
}
// Narrow — the segment shrinks every time, so this terminates.
if (inner > 0 && lastInner == segHi)
{
width = resolution; // a tie cluster at the top of the segment: read that one timestamp
lowerBy = 0;
bisecting = false;
}
else if (inner > 0 && (lowered || bisected))
{
// Data reaches past this segment's bottom: move `known` up to what this read saw; a gallop goes further.
known = lastInner;
if (lowered) lowerBy = lowerBy > maxWidth / 4 ? maxWidth : lowerBy * 4;
}
else if (inner > 0)
{
// Rows exist at lastInner; what lies above it is unknown. Aim at about half the page budget at the
// density the truncated rows showed (their span, not the segment's — it may open on an empty stretch).
var aimed = inner > 1 && lastInner > firstInner
? (long)Math.Min((double)(lastInner - firstInner) * ((useDiscoveryBudgetForGaps ? segBudget : pageBudget) / 2.0) / (inner - 1), maxWidth)
: resolution;
known = lastInner;
knownWidth = Math.Max(resolution, aimed);
width = Math.Max(resolution, Math.Min(aimed, segHi - segLo - 1));
lowerBy = 0;
bisecting = false;
}
else
{
// Only rows before the segment: shrink it — top-anchored from here on (a gallop would re-send the same
// crowded read).
width = Math.Max(resolution, Math.Min(width, segHi - segLo) / 4);
if (lowered || bisected) noGallop = true;
lowerBy = 0;
bisecting = false;
}
discovering = false;
narrowed = true;
break;
}
if (narrowed) continue; // re-read the same segHi, narrower
var count = AppendNewestFirst(rows, ticksOf, tieOrder, segLo, segHi, result, (int)Math.Min(int.MaxValue, want),
out var oldest);
// This complete read proves (segLo, oldest) empty — or the whole segment when it held nothing.
proven = count > 0 ? (segLo, oldest) : (segLo, segHi + 1);
if (probe)
{
probed = true;
segHi = segLo; // the whole remainder is read
continue;
}
emptyRun = count == 0 ? emptyRun + 1 : 0;
var grown = width > maxWidth / 4 ? maxWidth : width * 4;
if (count == 0)
{
discovering = true; // ×4 growth may overshoot into dense data
if (known > 0)
{
// An empty top-anchored segment: gallop up from `known`. An empty galloping read: the top is bracketed
// in (known, segLo] — bisect it (an empty bisecting read keeps bisecting).
if (lowered || bisected)
{
lowerBy = 0;
bisecting = true;
}
else if (noGallop)
{
width = maxWidth; // halve the gap, top-anchored
}
else
{
lowerBy = Math.Max(knownWidth, granularity);
}
}
else
{
width = grown;
}
}
else
{
// Re-aim at about half the page budget from the density just seen.
width = (long)Math.Clamp((double)(segHi - segLo) * Math.Max(1.0, pageBudget / 2.0) / count, resolution, grown);
discovering = false;
known = 0;
lowerBy = 0;
bisecting = false;
noGallop = false;
}
segHi = segLo;
if (known > 0 && segHi <= known)
{
// Everything above the newest row the truncated read saw is read: the next segment ends on it, at its density.
known = 0;
lowerBy = 0;
bisecting = false;
noGallop = false;
width = knownWidth;
discovering = false;
}
}
// Carry the stretch the last complete read proved empty, joined to the one carried in when they touch; when they
// do not, keep the one reaching the floor (it saves a whole walk), else the newer.
var next = carried;
if (proven.To > proven.From)
{
if (carried.Item2 > carried.Item1 && proven.From < carried.Item2 && carried.Item1 < proven.To)
next = (Math.Min(proven.From, carried.Item1), Math.Max(proven.To, carried.Item2));
else if (!(carried.Item2 > carried.Item1 && carried.Item1 <= stopTicks) || proven.From <= stopTicks)
next = proven;
}
return (result, new ReverseWalkHint(width, next.Item1, next.Item2));
}
/// <summary>Appends the rows in <c>(segLo, segHi]</c> newest-first (until <paramref name="limit"/>) and counts them.</summary>
private static int AppendNewestFirst<T>(
IReadOnlyList<T> rows, Func<T, long> ticksOf, IComparer<T>? tieOrder, long segLo, long segHi, List<T> result, int limit) =>
AppendNewestFirst(rows, ticksOf, tieOrder, segLo, segHi, result, limit, out _);
/// <summary>As above, also returning the oldest in-range timestamp (meaningless when none).</summary>
private static int AppendNewestFirst<T>(
IReadOnlyList<T> rows, Func<T, long> ticksOf, IComparer<T>? tieOrder, long segLo, long segHi, List<T> result, int limit,
out long oldest)
{
oldest = long.MaxValue;
IReadOnlyList<T> ordered = rows;
if (tieOrder is not null)
{
var copy = rows.ToList();
copy.Sort((a, b) =>
{
var c = ticksOf(a).CompareTo(ticksOf(b));
return c != 0 ? c : tieOrder.Compare(a, b);
});
ordered = copy;
}
var count = 0;
for (var i = ordered.Count - 1; i >= 0; i--)
{
var ts = ticksOf(ordered[i]);
if (ts <= segLo || ts > segHi) continue;
count++;
if (ts < oldest) oldest = ts;
if (result.Count < limit) result.Add(ordered[i]);
}
return count;
}
private static DateTime Utc(long ticks) => new(ticks, DateTimeKind.Utc);
private static long AlignDown(long ticks, long step) => ticks - ticks % step;
private static long AlignUp(long ticks, long step)
{
var down = AlignDown(ticks, step);
if (down == ticks) return ticks;
return down > DateTime.MaxValue.Ticks - step ? DateTime.MaxValue.Ticks : down + step;
}
}
/// <summary>
/// Orders events within one timestamp by <see cref="HistoricalEvent.EventId"/> (ordinal) — the tie order a walked-back
/// events read (#691.4) and its oversized-cluster over-fetch share, because the Gateway orders by event time only.
/// </summary>
internal sealed class EventIdTieOrder : IComparer<HistoricalEvent>
{
/// <summary>The shared instance.</summary>
public static readonly EventIdTieOrder Instance = new();
/// <inheritdoc />
public int Compare(HistoricalEvent? x, HistoricalEvent? y) => string.CompareOrdinal(x?.EventId, y?.EventId);
}