lmxopcua/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/GatewayHistorianDataSource.cs

using System.Collections.Frozen;
using Grpc.Core;
using Microsoft.Extensions.Logging;
using ZB.MOM.WW.HistorianGateway.Contracts.Grpc;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Mapping;
using ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common;
using ZB.MOM.WW.OtOpcUa.Runtime.Historian;

namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway;

/// <summary>
///     Server-side <see cref="IHistorianDataSource"/> backed by the HistorianGateway gRPC surface
///     (via the <see cref="IHistorianGatewayClient"/> seam). Translates OPC UA HistoryRead requests
///     to gateway read calls and maps the wire shapes back to the driver-agnostic
///     <see cref="DataValueSnapshot"/> / <see cref="HistoricalEvent"/> carriers using the pure
///     mappers in <c>Mapping/</c>.
/// </summary>
/// <remarks>
///     The data source owns no historian connection of its own — it delegates to the gateway, which
///     pools and amortizes the underlying historian sessions. A thrown gateway exception is recorded
///     as a health failure and rethrown: the node manager turns it into a Bad HistoryRead result, so
///     a backend fault never crashes the host. An empty time window is a successful (GoodNoData)
///     read, not a fault. Health counters follow the single-<c>_healthLock</c> discipline ported
///     from <c>WonderwareHistorianClient</c> so <c>TotalSuccesses + TotalFailures == TotalQueries</c>
///     holds at every observed snapshot.
/// </remarks>
public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDisposable
{
    /// <summary>
    ///     The processed (aggregate) functions the Gateway backend serves — the single source of truth for this backend's
    ///     aggregate set: the per-sample aggregates evaluated from raw reads (every member of the
    ///     <see cref="AggregateFamily.Simple"/> / <see cref="AggregateFamily.StartEnd"/> /
    ///     <see cref="AggregateFamily.ActualTime"/> families of <see cref="AggregateFamilies"/>: Average, Count, Minimum,
    ///     Maximum, Range, Start, End, Delta, MinimumActualTime, MaximumActualTime) plus the two answered natively by AVEVA
    ///     (TimeAverage, Total — <see cref="AggregateModeMapper"/>). Any other aggregate throws
    ///     <see cref="HistoryAggregateNotSupportedException"/> without calling the gateway. It is what
    ///     <see cref="SupportedAggregates"/> returns, so the advertised <c>AggregateFunctions</c> folders and the server's
    ///     up-front <c>BadAggregateNotSupported</c> read it too.
    /// </summary>
    /// <remarks>
    ///     The two native members are served only with <c>ServerHistorian:Gateway:EnableNativeTimeWeighted</c> (default
    ///     <c>true</c> since lmxopcua#694, once the gateway's interval fix historiangw#27 was live-verified); with it off an
    ///     instance serves (and advertises) <see cref="RawEvaluatedAggregates"/> only — see <see cref="SupportedAggregates"/>.
    /// </remarks>
    public static IReadOnlySet<HistoryAggregateType> ServedAggregates { get; } =
        Enum.GetValues<HistoryAggregateType>()
            .Where(a => RawAggregateReader.Serves(a) || AggregateModeMapper.NativeAggregates.Contains(a))
            .ToFrozenSet();

    /// <summary>
    ///     The per-sample aggregates this backend evaluates from raw reads — what an instance serves with
    ///     <c>ServerHistorian:Gateway:EnableNativeTimeWeighted=false</c> (the kill switch).
    /// </summary>
    public static IReadOnlySet<HistoryAggregateType> RawEvaluatedAggregates { get; } =
        Enum.GetValues<HistoryAggregateType>().Where(RawAggregateReader.Serves).ToFrozenSet();

    /// <summary>Whether AVEVA's native TimeAverage / Total are served (opt-in; see <see cref="ServedAggregates"/>).</summary>
    private readonly bool _enableNativeTimeWeighted;

    private readonly IHistorianGatewayClient _client;
    private readonly ILogger<GatewayHistorianDataSource> _logger;
    private readonly int _maxRawRows;
    private readonly TimeProvider _time;
    private readonly bool _atTimeFromRaw;

    /// <summary>AtTime from raw reads (true) or AVEVA's own interpolation (the kill switch).</summary>
    internal bool AtTimeFromRaw => _atTimeFromRaw;

    /// <summary>The most crowding re-reads one raw read makes: each at least doubles the pre-window room.</summary>
    private const int MaxCrowdedReReads = 8;

    /// <summary>The default of <c>maxPreWindowRows</c>: <c>ServerHistorian:MaxTieClusterOverfetch</c>'s default.</summary>
    public const int DefaultMaxPreWindowRows = 65536;

    /// <summary>
    ///     The most rows before a raw window (inside the 1 ms look-back) a crowding re-read makes room for — the
    ///     server's tie-cluster limit (<c>ServerHistorian:MaxTieClusterOverfetch</c>), which every other tie path treats as
    ///     the limit. More ⇒ <see cref="HistoryReadLimitExceededException"/> with <c>PreWindowRows</c> set.
    /// </summary>
    private readonly int _maxPreWindowRows;

    /// <summary>The tie-cluster limit the crowding re-reads honour — exposed so the factory wiring is testable.</summary>
    internal int MaxPreWindowRows => _maxPreWindowRows;

    /// <summary>
    ///     The server raw cap (<c>ServerHistorian:MaxRawValuesPerRead</c>, R08:1.1): the most samples one raw read may
    ///     return; 0 = none. See <see cref="MaxRawValuesPerRead"/> and <see cref="ReadRawWindowAsync"/>.
    /// </summary>
    private readonly int _maxRawValuesPerRead;

    /// <summary>The default of <c>eventDrainBudget</c>: <c>ServerHistorian:Gateway:EventDrainBudget</c>'s default.</summary>
    public const int DefaultEventDrainBudget = GatewayHistorianSettings.DefaultEventDrainBudget;

    /// <summary>
    ///     Stream rows an events read drains past its cap before it abandons the gateway stream; see
    ///     <see cref="ReadEventsAsync(string?, DateTime, DateTime, int, CancellationToken)"/>.
    /// </summary>
    private readonly int _eventDrainBudget;

    private long _eventStreamsAbandoned;

    /// <summary>The bound events drain budget — exposed so the factory wiring is testable.</summary>
    internal int EventDrainBudget => _eventDrainBudget;

    /// <summary>Events reads that abandoned their gateway stream past the drain budget (lifetime) — test/diagnostic.</summary>
    internal long EventStreamsAbandoned => Interlocked.Read(ref _eventStreamsAbandoned);

    /// <summary>A strict gateway's row ceiling, learned from its first refusal.</summary>
    private readonly GatewayRowCeiling _ceiling;
    private readonly GatewayTagPresenceCache _tags;

    private readonly object _healthLock = new();
    private DateTime? _lastSuccessUtc;
    private DateTime? _lastFailureUtc;
    private string? _lastError;
    private long _totalQueries;
    private long _totalSuccesses;
    private long _totalFailures;
    private int _consecutiveFailures;

    /// <summary>
    ///     The latest connectivity observation — a read outcome or a probe result, whichever came last (null = none yet).
    ///     <see cref="GetHealthSnapshot"/> reports it as the connection flags.
    /// </summary>
    private bool? _lastObservationOpen;

    /// <summary>The probe cadence (<c>ServerHistorian:Gateway:ProbeInterval</c>); zero = no probe loop.</summary>
    private readonly TimeSpan _probeInterval;

    /// <summary>The probe cadence — exposed so the factory wiring is testable.</summary>
    internal TimeSpan ProbeInterval => _probeInterval;

    private readonly CancellationTokenSource _probeStop = new();
    private readonly Task _probeLoop = Task.CompletedTask;

    /// <summary>The most one probe may take before it counts as a failed (closed) observation.</summary>
    private static readonly TimeSpan ProbeCallBound = TimeSpan.FromSeconds(10);

    /// <summary>Creates a gateway-backed historian data source.</summary>
    /// <param name="client">The gateway client seam used for all reads.</param>
    /// <param name="logger">Diagnostic logger; failures are recorded without leaking tag/host detail.</param>
    /// <param name="maxRawRows">
    ///     In-window raw-row cap of one raw-evaluated processed read (<c>ServerHistorian:MaxTimeWeightedRawRows</c>;
    ///     &lt;= 0 = unbounded). Past it the node fails <c>BadTooManyOperations</c>.
    /// </param>
    /// <param name="timeProvider">Clock for result server timestamps and the raw look-ahead horizon; null = system.</param>
    /// <param name="atTimeFromRaw">
    ///     serve AtTime from raw reads with the shared AtTime rules (<see cref="GatewayAtTimeReader"/>; default);
    ///     <c>false</c> = AVEVA's own interpolation passed through (<c>ServerHistorian:GatewayAtTimeFromRaw</c> kill switch).
    /// </param>
    /// <param name="maxPreWindowRows">
    ///     The most rows before a raw window a crowding re-read makes room for (<c>ServerHistorian:MaxTieClusterOverfetch</c>,
    ///     values below 1 count as 1). More ⇒ <see cref="HistoryReadLimitExceededException"/> (<c>BadTooManyOperations</c>).
    /// </param>
    /// <param name="maxRawValuesPerRead">
    ///     The server raw cap (<c>ServerHistorian:MaxRawValuesPerRead</c>, R08:1.1): the most samples one
    ///     <see cref="ReadRawAsync"/> may return — reported as <see cref="MaxRawValuesPerRead"/> (bounded further by a strict
    ///     gateway's learned ceiling), so the server pages at it. A read asking for more (or for everything) asks the gateway
    ///     for cap + 1 rows and is refused when the window holds more, never truncated. &lt;= 0 = unbounded (the pre-#736
    ///     behaviour: an uncapped read is sent as <see cref="int.MaxValue"/>).
    /// </param>
    /// <param name="enableNativeTimeWeighted">
    ///     Serve TimeAverage / Total through AVEVA's native retrieval modes (<c>ServerHistorian:Gateway:EnableNativeTimeWeighted</c>,
    ///     default <c>true</c> since lmxopcua#694). <c>false</c> refuses both <see cref="HistoryAggregateNotSupportedException"/>
    ///     (<c>BadAggregateNotSupported</c>) and stops advertising them — for a gateway older than the historiangw#27 fix.
    /// </param>
    /// <param name="probeInterval">
    ///     Cadence of the <c>ProbeAsync</c> health loop (<c>ServerHistorian:Gateway:ProbeInterval</c>); the first probe runs
    ///     at construction. Null / zero (the default here; the factory passes the configured 30 s) = no loop, health from
    ///     read outcomes only.
    /// </param>
    /// <param name="eventDrainBudget">
    ///     Stream rows a capped events read consumes (uncollected) past its cap before it abandons the gateway stream
    ///     (<c>ServerHistorian:Gateway:EventDrainBudget</c>). Negative counts as 0 = abandon at the cap.
    /// </param>
    public GatewayHistorianDataSource(IHistorianGatewayClient client, ILogger<GatewayHistorianDataSource> logger,
        int maxRawRows = TimeWeightedAggregator.DefaultMaxWindowRows, TimeProvider? timeProvider = null,
        bool atTimeFromRaw = true, int maxPreWindowRows = DefaultMaxPreWindowRows, int maxRawValuesPerRead = 0,
        int eventDrainBudget = DefaultEventDrainBudget, bool enableNativeTimeWeighted = true,
        TimeSpan? probeInterval = null)
    {
        _enableNativeTimeWeighted = enableNativeTimeWeighted;
        _eventDrainBudget = Math.Max(0, eventDrainBudget);
        _maxPreWindowRows = Math.Max(1, maxPreWindowRows);
        _maxRawValuesPerRead = Math.Max(0, maxRawValuesPerRead);
        _atTimeFromRaw = atTimeFromRaw;
        ArgumentNullException.ThrowIfNull(client);
        ArgumentNullException.ThrowIfNull(logger);
        _client = client;
        _logger = logger;
        _maxRawRows = maxRawRows;
        _time = timeProvider ?? TimeProvider.System;
        _ceiling = new GatewayRowCeiling(_time);
        _tags = new GatewayTagPresenceCache(_client, _time);
        _probeInterval = probeInterval is { } pi && pi > TimeSpan.Zero ? pi : TimeSpan.Zero;
        if (_probeInterval > TimeSpan.Zero) _probeLoop = Task.Run(() => ProbeLoopAsync(_probeStop.Token));
    }

    /// <summary>
    ///     Probes the gateway once immediately, then every <see cref="_probeInterval"/>, recording each result as the
    ///     latest connectivity observation. Probes are not history queries: they never touch the query / failure counters.
    ///     A probe that throws or overruns <see cref="ProbeCallBound"/> is a closed observation; nothing escapes the loop.
    /// </summary>
    private async Task ProbeLoopAsync(CancellationToken stop)
    {
        using var timer = new PeriodicTimer(_probeInterval, _time);
        do
        {
            bool open;
            try
            {
                using var bound = CancellationTokenSource.CreateLinkedTokenSource(stop);
                bound.CancelAfter(ProbeCallBound);
                open = await _client.ProbeAsync(bound.Token).ConfigureAwait(false);
            }
            catch (OperationCanceledException) when (stop.IsCancellationRequested)
            {
                return;
            }
            catch (Exception ex)
            {
                open = false;
                _logger.LogDebug("Historian gateway probe failed ({Exception}).", ex.GetType().Name);
            }

            bool changed;
            lock (_healthLock)
            {
                changed = _lastObservationOpen != open;
                _lastObservationOpen = open;
            }
            if (changed)
                _logger.LogInformation("Historian gateway probe: the gateway is {State}.", open ? "reachable" : "unreachable");
        }
        while (await WaitNextAsync(timer, stop).ConfigureAwait(false));
    }

    private static async Task<bool> WaitNextAsync(PeriodicTimer timer, CancellationToken stop)
    {
        try { return await timer.WaitForNextTickAsync(stop).ConfigureAwait(false); }
        catch (OperationCanceledException) { return false; }
    }

    /// <inheritdoc />
    // A reverse window (start > end) is refused with NotSupportedException before anything is sent
    // and is not recorded as a health failure: the gateway rejects it InvalidArgument, and the server
    // never asks — SupportsReverseRaw is false, so it walks a reverse read back with forward reads.
    // Stored samples only (Part 11 Raw): AVEVA's synthetic start-boundary row is never returned, see
    // ReadRawWindowAsync.
    public async Task<HistoryReadResult> ReadRawAsync(
        string fullReference, DateTime startUtc, DateTime endUtc, uint maxValuesPerNode,
        CancellationToken cancellationToken)
    {
        if (startUtc > endUtc)
            throw new NotSupportedException("The HistorianGateway reads raw history oldest-first only (start > end).");

        try
        {
            if (!await _tags.ExistsAsync(fullReference, cancellationToken).ConfigureAwait(false))
            {
                RecordOutcome(success: true, error: null);
                return new HistoryReadResult([], ContinuationPoint: null);
            }
            var samples = await ReadRawWindowAsync(fullReference, startUtc, endUtc, maxValuesPerNode, cancellationToken)
                .ConfigureAwait(false);
            var snapshots = SampleMapper.ToSnapshots(samples);
            RecordOutcome(success: true, error: null);
            return new HistoryReadResult(snapshots, ContinuationPoint: null);
        }
        catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
        {
            throw; // the caller (client cancel / HistoryReadDeadline) asked for this — not a backend fault
        }
        catch (Exception ex) when (IsMaxValuesCeilingRejection(ex))
        {
            // An unbounded read over Historian:ReadLimits:MaxValuesCeiling with AllowStreamingBeyondCeiling=false: a
            // client-request limit (BadTooManyOperations), not a backend fault.
            throw new HistoryReadLimitExceededException(maxValuesPerNode,
                $"The HistorianGateway refused the row budget of the raw read of '{fullReference}' (its " +
                "Historian:ReadLimits:MaxValuesCeiling); set NumValuesPerNode or narrow the window.");
        }
        catch (HistoryReadLimitExceededException)
        {
            // A client-request limit (the pre-window margin swallowed the row budget), not a backend fault.
            throw;
        }
        catch (Exception ex)
        {
            RecordReadFailure(ex);
            if (IsTransportFailure(ex)) throw new System.IO.IOException("HistorianGateway read transport is unavailable.", ex);
            throw;
        }
    }

    /// <summary>
    ///     The stored samples of <c>[startUtc, endUtc)</c> (a point read <c>start == end</c>: of that instant), oldest first,
    ///     at most <paramref name="maxValuesPerNode"/> (<c>0</c> = all) — without AVEVA's synthetic rows.
    /// </summary>
    /// <remarks>
    ///     <para>
    ///         An AVEVA <c>Full</c> raw read starts with a row stamped exactly at the query start: the value in effect there
    ///         (<c>Quality = 133</c>, the prior sample's value — never stored at that time) or, before the tag's first sample,
    ///         a NULL row (value 0, Bad). Part 11 Raw with <c>ReturnBounds = false</c> wants stored values only, and the
    ///         server's paging counts every returned row toward the page and resumes at the last returned timestamp,
    ///         so a fabricated row would also be counted as a sample and re-fabricated at every resume point. The read
    ///         therefore uses the same request start as the raw-evaluated aggregates: it is requested from
    ///         <see cref="RawAggregateReader.RequestStartFor"/> (1 ms early) and only rows
    ///         <see cref="RawAggregateReader.IsWindowSample"/> are kept — the decision is by TIMESTAMP, because a stored NULL
    ///         sample and the no-data boundary row look the same on the wire (both carry a value 0 with a Bad quality). The
    ///         forward end is exclusive (the interface contract; the gateway's own end is inclusive), a point read inclusive.
    ///     </para>
    ///     <para>
    ///         <b>Row budget.</b> <c>0</c> ("all values") asks for <see cref="int.MaxValue"/> (the client package and the
    ///         gateway reject <c>max_values &lt;= 0</c>). A cap asks for <c>cap + </c>
    ///         <see cref="RawAggregateReader.PreWindowAllowance"/> rows — NOT clamped at the gateway's default ceiling (this
    ///         clamps only the per-sample aggregate budget): AVEVA always emits the boundary row, so a clamped budget would make
    ///         every page size within the allowance of the ceiling come back short. A strict gateway
    ///         (<c>AllowStreamingBeyondCeiling=false</c>) refuses an over-ceiling budget and names its ceiling; the read is
    ///         then re-issued ONCE with the ceiling as budget (<see cref="GatewayReadLimits"/>) — so an uncapped read
    ///         of a small window is served — and a ceiling-bounded read that is not provably complete (it delivered the
    ///         whole ceiling without reaching the window end, or a page short of its cap) fails
    ///         <see cref="HistoryReadLimitExceededException"/> (<c>BadTooManyOperations</c>), never a truncated answer.
    ///     </para>
    ///     <para>
    ///         <b>Completeness.</b> A full page must hold exactly the cap of stored samples — a short page ends the client's
    ///         paging. A read is complete when the stream came back short, filled the cap, or reached the window end. If rows
    ///         BEFORE the window (the boundary row and stored samples inside the 1 ms margin, e.g. a tie cluster at
    ///         <c>start − 1 ms</c>) used up the budget, the window is read ONCE more asking for <c>cap + m + </c>
    ///         <see cref="RawAggregateReader.PreWindowAllowance"/> (m = the pre-window rows the first read saw); only if that
    ///         re-read is still short because of pre-window rows does the read fail with
    ///         <see cref="HistoryReadLimitExceededException"/>. Every stream is consumed to completion (never broken out of —
    ///         a cancelled gateway stream evicts its pooled historian session). <c>ReturnBounds</c> is not plumbed to the
    ///         data source on any backend, so bounding values are never synthesised here.
    ///     </para>
    ///     <para>
    ///         <b>The server raw cap</b> (R08:1.1, #736). With a cap, a read asking for everything (<c>0</c>) or for more than
    ///         the cap runs as a capped read of cap + 1 — one more than it may return — so an over-cap window is DETECTED and
    ///         refused (<see cref="HistoryReadLimitExceededException"/>) rather than answered truncated (the server's paging
    ///         reads a short answer as the last page) and never sent as <see cref="int.MaxValue"/>. The server itself never asks
    ///         for more than the cap (it pages at <see cref="MaxRawValuesPerRead"/>); this bounds a caller that bypasses it.
    ///     </para>
    /// </remarks>
    private async Task<List<HistorianSample>> ReadRawWindowAsync(
        string fullReference, DateTime startUtc, DateTime endUtc, uint maxValuesPerNode,
        CancellationToken cancellationToken)
    {
        // The gateway seam caps with an int; OPC UA hands us a uint, so clamp to int range.
        var cap = (int)Math.Min(maxValuesPerNode, int.MaxValue);
        var serverCap = _maxRawValuesPerRead;
        // Under the server raw cap, a read asking for everything or for more than the cap runs as a capped read of
        // cap + 1 (budgetCap); `cap` stays what the client asked for, and a refusal names the cap that applied (limit).
        var probe = serverCap > 0 && (cap == 0 || cap > serverCap);
        var budgetCap = probe ? (int)Math.Min(int.MaxValue, (long)serverCap + 1) : cap;
        var limit = probe ? serverCap : cap;
        var advice = probe
            ? $"set NumValuesPerNode at or below ServerHistorian:MaxRawValuesPerRead ({serverCap}) to page"
            : cap == 0 ? "set NumValuesPerNode to page" : "lower NumValuesPerNode";
        if (budgetCap == 0)
        {
            var all = await PassAsync(int.MaxValue).ConfigureAwait(false);
            if (all.Result.Complete) return all.Result.Samples;
            // Only reachable on a strict gateway: the ceiling re-read delivered the whole ceiling without reaching
            // the window end — the window may hold more, and "all values" is never answered from a truncated read.
            throw new HistoryReadLimitExceededException(limit,
                $"The raw read of '{fullReference}' holds more than the HistorianGateway's row ceiling ({all.Budget}, its " +
                "Historian:ReadLimits:MaxValuesCeiling with AllowStreamingBeyondCeiling=false); set NumValuesPerNode to page, " +
                "or narrow the window.")
            { PreWindowRows = Crowding(all.Result.PreWindowRows) };
        }

        var asked = (int)Math.Min(int.MaxValue, (long)budgetCap + RawAggregateReader.PreWindowAllowance);
        var pass = await PassAsync(asked).ConfigureAwait(false);
        if (pass.Result.Complete) return Bounded(pass.Result.Samples);

        // A strict gateway's ceiling already bounded that read: a larger budget would only be refused again. The
        // refusal carries the pre-window rows it saw: the server re-reads the page that many values smaller.
        if (pass.Budget < asked)
            throw new HistoryReadLimitExceededException(limit,
                $"The raw read of '{fullReference}' could not be read completely within the HistorianGateway's row ceiling " +
                $"({pass.Budget}, its Historian:ReadLimits:MaxValuesCeiling with AllowStreamingBeyondCeiling=false); {advice}, " +
                "or narrow the window.")
            { PreWindowRows = Crowding(pass.Result.PreWindowRows) };

        // The margin crowded the window out: re-read with room for the pre-window rows. A read that was itself cut short saw
        // only some of them (a tie cluster of hundreds just before a resume point used to fail even with no
        // ceiling), so the room at least doubles per re-read — bounded by MaxCrowdedReReads, and never past a known ceiling.
        // The room is capped at the tie-cluster limit (MaxTieClusterOverfetch) — a pathological single-millisecond
        // cluster on a streaming gateway would otherwise be streamed several times over for one page.
        var preWindow = pass.Result.PreWindowRows;
        var room = Math.Min(Math.Max(1, preWindow), _maxPreWindowRows);
        for (var attempt = 0; attempt < MaxCrowdedReReads && preWindow <= _maxPreWindowRows; attempt++)
        {
            var budget = (int)Math.Min(int.MaxValue, (long)budgetCap + room + RawAggregateReader.PreWindowAllowance);
            // doubling can leap over the ceiling while the cluster still fits. Spend the remaining room once.
            if (_ceiling.Learned is var known and > 0 && budget > known)
            {
                var available = known - budgetCap - RawAggregateReader.PreWindowAllowance;
                if (available <= 0) break;
                room = Math.Min(room, available);
                budget = budgetCap + room + RawAggregateReader.PreWindowAllowance;
            }
            if (budget <= pass.Budget) break; // no more room, never retry the same full read
            pass = await PassAsync(budget).ConfigureAwait(false);
            if (pass.Result.Complete) return Bounded(pass.Result.Samples);
            preWindow = Math.Max(preWindow, pass.Result.PreWindowRows);
            if (pass.Budget < budget) break; // bounded by a ceiling learned meanwhile
            if (room >= _maxPreWindowRows) break; // the limit's worth of room did not suffice
            room = (int)Math.Min(_maxPreWindowRows, Math.Max(preWindow, 2L * room));
        }

        throw new HistoryReadLimitExceededException(limit,
            $"The raw read of '{fullReference}' could not be read completely: more than {pass.Result.PreWindowRows} rows lie in " +
            $"the {RawAggregateReader.BoundaryMargin.TotalMilliseconds} ms before the window and exhausted the row budget.")
        { PreWindowRows = preWindow };

        // Pre-window rows the allowance did not cover are what crowded the read; fewer are not the cause.
        static int Crowding(int preWindowRows) => preWindowRows > RawAggregateReader.PreWindowAllowance ? preWindowRows : 0;

        // The cap + 1 probe came back with one sample too many: the window holds more than the server raw cap.
        List<HistorianSample> Bounded(List<HistorianSample> samples)
        {
            if (probe && samples.Count > serverCap)
                throw new HistoryReadLimitExceededException(serverCap,
                    $"The raw read of '{fullReference}' holds more than ServerHistorian:MaxRawValuesPerRead ({serverCap}) " +
                    "samples; set NumValuesPerNode at or below it to page, or narrow the window.");
            return samples;
        }

        // One drained pass; a strict gateway's refusal of the budget is retried once at the ceiling it names.
        Task<((List<HistorianSample> Samples, int PreWindowRows, bool Complete) Result, int Budget)> PassAsync(int budget) =>
            _ceiling.ReadAsync(budget,
                b => ReadRawPassAsync(fullReference, startUtc, endUtc, budgetCap, b, cancellationToken),
                (r, _) => !r.Complete);
    }

    /// <summary>One forward raw read of a window, drained, with its pre-window row count and whether it is complete.</summary>
    private async Task<(List<HistorianSample> Samples, int PreWindowRows, bool Complete)> ReadRawPassAsync(
        string fullReference, DateTime startUtc, DateTime endUtc, int cap, int budget, CancellationToken cancellationToken)
    {
        var samples = new List<HistorianSample>();
        long received = 0;
        var preWindow = 0;
        var reachedEnd = false;
        await foreach (var sample in _client
            .ReadRawAsync(fullReference, RawAggregateReader.RequestStartFor(startUtc), endUtc, budget, cancellationToken)
            .ConfigureAwait(false))
        {
            received++;
            // Filter while draining — the rest of the (bounded) stream is consumed, never abandoned.
            if (RawAggregateReader.IsWindowSample(sample, startUtc, endUtc))
            {
                if (cap == 0 || samples.Count < cap) samples.Add(sample);
            }
            // A row before the window — or, for a window starting at/before the epoch (no look-back), AVEVA's synthetic row at
            // the query start, which is no sample either (it used to set reachedEnd, so a ceiling-truncated read
            // passed as complete) — is a pre-window row.
            else if (sample.Timestamp is { } ts
                     && (ts.ToDateTime() < startUtc || !RawAggregateReader.IsAfterFloor(ts.ToDateTime()))) preWindow++;
            else if (sample.Timestamp is not null) reachedEnd = true;
        }

        // cap == 0 ("all values") is complete only when the stream proves it: short of the budget (always, unless a strict
        // gateway's ceiling became the budget) or past the window end.
        var complete = received < budget || reachedEnd || (cap > 0 && samples.Count >= cap);
        return (samples, preWindow, complete);
    }

    /// <inheritdoc />
    //
    // Per-sample aggregates — Average, Count, Minimum, Maximum, Range, Start, End, Delta,
    // Minimum/MaximumActualTime — are evaluated from a raw read with the SQL backends' shared evaluator
    // (RawAggregateReader): identical values, status, Calculated/Partial bits and timestamps to
    // SqlServer/QuestDB for the same samples, forward and reverse. The in-window raw rows are capped by
    // ServerHistorian:MaxTimeWeightedRawRows (HistoryReadLimitExceededException, not counted as
    // a backend failure).
    //
    // TimeAverage / Total stay native AVEVA retrieval modes (AggregateModeMapper), stamped with
    // the interval start and the Calculated bit; AVEVA's extra row for the interval before the window is dropped and
    // an Uncertain interval keeps its value. Both directions use the SQL backends' interval grid with forward
    // requests only — a reverse read (which the gateway refuses) anchored at its later time, the clipped interval
    // read over exactly its span and marked Partial (NativeAggregatePlan). Everything else is HistoryAggregateNotSupportedException
    // (see ServedAggregates).
    //
    // A raw-evaluated read whose window holds more than BucketPlan.MaxBuckets intervals throws.
    public async Task<HistoryReadResult> ReadProcessedAsync(
        string fullReference, DateTime startUtc, DateTime endUtc, TimeSpan interval,
        HistoryAggregateType aggregate, CancellationToken cancellationToken)
    {
        if (RawAggregateReader.Serves(aggregate))
            return await ReadProcessedFromRawAsync(fullReference, startUtc, endUtc, interval, aggregate, cancellationToken)
                .ConfigureAwait(false);

        // Mapped OUTSIDE the failure-recording try: an aggregate the gateway has no mode for is a client
        // request the server answers BadAggregateNotSupported, not a backend fault to count. The native pair is refused
        // the same way when EnableNativeTimeWeighted is off (the kill switch for a gateway older than historiangw#27).
        var mode = AggregateModeMapper.ToRetrievalMode(aggregate);
        if (!_enableNativeTimeWeighted) throw new HistoryAggregateNotSupportedException(aggregate);
        // Planned OUTSIDE the failure-recording try: too many intervals is a client request limit, not a backend fault.
        return await ReadNativeAsync(fullReference, BucketPlan.Create(startUtc, endUtc, interval), mode, cancellationToken)
            .ConfigureAwait(false);
    }

    /// <summary>
    ///     A native TimeAverage / Total read on the SQL backends' interval grid, either direction: forward native
    ///     requests only, the clipped interval read over exactly its span. See <see cref="NativeAggregatePlan"/>.
    /// </summary>
    private async Task<HistoryReadResult> ReadNativeAsync(string fullReference, BucketPlan plan, RetrievalMode mode,
        CancellationToken cancellationToken)
    {
        if (plan.BucketCount == 0) return new HistoryReadResult([], ContinuationPoint: null);
        // Planned OUTSIDE the failure-recording try: a non-whole-ms interval over too many intervals is a client request limit.
        var requests = NativeAggregatePlan.Requests(plan);
        try
        {
            if (!await _tags.ExistsAsync(fullReference, cancellationToken).ConfigureAwait(false))
            {
                RecordOutcome(success: true, error: null);
                return new HistoryReadResult(plan.Assemble(new Dictionary<long, DataValueSnapshot>(), t => AggregateEvaluator.NoData(t, _time.GetUtcNow().UtcDateTime)), ContinuationPoint: null);
            }
            var byBucket = new Dictionary<long, DataValueSnapshot>();
            foreach (var request in requests)
            {
                var buckets = new List<HistorianAggregateSample>();
                await foreach (var bucket in _client
                    .ReadAggregateAsync(fullReference, request.StartUtc, request.EndUtc, mode, request.Interval, cancellationToken)
                    .ConfigureAwait(false))
                {
                    buckets.Add(bucket); // drained to completion, mapped afterwards
                }
                NativeAggregatePlan.Place(plan, request, buckets, byBucket);
            }

            RecordOutcome(success: true, error: null);
            return new HistoryReadResult(NativeAggregatePlan.Assemble(plan, byBucket), ContinuationPoint: null);
        }
        catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
        {
            throw; // caller-requested, not a backend fault
        }
        catch (Exception ex)
        {
            RecordReadFailure(ex);
            if (IsTransportFailure(ex)) throw new System.IO.IOException("HistorianGateway read transport is unavailable.", ex);
            throw;
        }
    }

    /// <summary>Serves a per-sample aggregate from a raw read; see <see cref="RawAggregateReader"/>.</summary>
    private async Task<HistoryReadResult> ReadProcessedFromRawAsync(
        string fullReference, DateTime startUtc, DateTime endUtc, TimeSpan interval,
        HistoryAggregateType aggregate, CancellationToken cancellationToken)
    {
        // Planned OUTSIDE the failure-recording try (as on the SQL backends): too many intervals is a client request
        // limit, not a backend fault.
        var plan = BucketPlan.Create(startUtc, endUtc, interval);
        try
        {
            var now = _time.GetUtcNow().UtcDateTime;
            if (plan.BucketCount > 0 && !await _tags.ExistsAsync(fullReference, cancellationToken).ConfigureAwait(false))
            {
                RecordOutcome(success: true, error: null);
                return new HistoryReadResult(plan.Assemble(new Dictionary<long, DataValueSnapshot>(), t => AggregateEvaluator.NoData(t, now)), ContinuationPoint: null);
            }
            var values = await RawAggregateReader
                .ReadAsync(_client, _ceiling, fullReference, aggregate, plan, _maxRawRows, now, cancellationToken)
                .ConfigureAwait(false);
            RecordOutcome(success: true, error: null);
            return new HistoryReadResult(values, ContinuationPoint: null);
        }
        catch (HistoryReadLimitExceededException)
        {
            // A client-request limit (the raw-row cap), not a backend fault: not counted as a failure.
            throw;
        }
        catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
        {
            // The caller (client cancel / HistoryReadDeadline) asked for this — not a backend fault.
            throw;
        }
        catch (Exception ex) when (IsMaxValuesCeilingRejection(ex))
        {
            // The gateway refused the row budget (Historian:ReadLimits:MaxValuesCeiling with
            // AllowStreamingBeyondCeiling=false): the same client-request limit as the cap, not a backend fault.
            throw new HistoryReadLimitExceededException(_maxRawRows,
                $"The HistorianGateway refused the raw row budget of the processed read of '{fullReference}' (its " +
                "Historian:ReadLimits:MaxValuesCeiling); lower ServerHistorian:MaxTimeWeightedRawRows below that ceiling " +
                "or narrow the window.");
        }
        catch (Exception ex)
        {
            RecordReadFailure(ex);
            if (IsTransportFailure(ex)) throw new System.IO.IOException("HistorianGateway read transport is unavailable.", ex);
            throw;
        }
    }

    /// <summary>
    ///     True when <paramref name="ex"/> (or an inner exception — the client package wraps the <see cref="RpcException"/>)
    ///     is the gateway's <c>InvalidArgument</c> "max_values exceeds the server ceiling" rejection.
    /// </summary>
    /// <param name="ex">The exception to classify.</param>
    /// <returns><see langword="true"/> when the exception is the gateway's row-ceiling rejection.</returns>
    internal static bool IsMaxValuesCeilingRejection(Exception ex) => GatewayReadLimits.IsCeilingRejection(ex);

    /// <inheritdoc />
    // served from RAW gateway reads — each requested time's stored sample at-or-before and after it — evaluated with
    // the SQL backends' shared AtTimeEvaluator (Part 11 §6.5.4), so the Gateway answers what SqlServer and
    // QuestDB answer for the same stored samples; see GatewayAtTimeReader for the read plan and its cost.
    // AVEVA's own interpolated answer (the gateway's ReadAtTime) returned the previous value as Good at a stored
    // NULL's time, null + Bad at a Good sample right after a NULL, a drifted value at an exact stored sample, and no
    // Interpolated bit; it is only used when ServerHistorian:GatewayAtTimeFromRaw = false (kill switch).
    // A gateway value is a Double (or a String) whatever the node's type (AVEVA stores a Boolean as a 0/1 Int2), so
    // without a type hint every numeric tag is interpolated linearly; the node manager passes the node's type through
    // ReadAtTimeAsync(string, IReadOnlyList<DateTime>, TypeCode?, CancellationToken).
    public Task<HistoryReadResult> ReadAtTimeAsync(
        string fullReference, IReadOnlyList<DateTime> timestampsUtc, CancellationToken cancellationToken) =>
        ReadAtTimeAsync(fullReference, timestampsUtc, valueType: null, cancellationToken);

    /// <inheritdoc />
    // the hint makes the AtTime rules type-exact, as on the SQL backends — the gateway's Double rows are converted to
    // the node's type BEFORE the shared evaluator runs (Boolean = value ≠ 0 ⇒ stepped; an integer type ⇒ Int64 ⇒
    // interpolated and rounded half away from zero, then returned as the node's integer type; String ⇒ stepped; Float /
    // Double / no hint ⇒ linear Double). Ignored with the kill switch (AVEVA's own answer).
    public async Task<HistoryReadResult> ReadAtTimeAsync(
        string fullReference, IReadOnlyList<DateTime> timestampsUtc, TypeCode? valueType, CancellationToken cancellationToken)
    {
        ArgumentNullException.ThrowIfNull(timestampsUtc);
        if (!_atTimeFromRaw) return await ReadAtTimeNativeAsync(fullReference, timestampsUtc, cancellationToken).ConfigureAwait(false);
        try
        {
            var now = _time.GetUtcNow().UtcDateTime;
            if (timestampsUtc.Count > 0 && !await _tags.ExistsAsync(fullReference, cancellationToken).ConfigureAwait(false))
            {
                RecordOutcome(success: true, error: null);
                return MissingAtTime(timestampsUtc, now);
            }
            var values = await GatewayAtTimeReader
                .ReadAsync(_client, _ceiling, fullReference, timestampsUtc, now, cancellationToken, valueType)
                .ConfigureAwait(false);
            RecordOutcome(success: true, error: null);
            return new HistoryReadResult(values, ContinuationPoint: null);
        }
        catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
        {
            throw; // caller-requested, not a backend fault
        }
        catch (HistoryReadLimitExceededException)
        {
            throw; // a client-request limit (a pathological tie cluster), not a backend fault
        }
        catch (Exception ex)
        {
            RecordReadFailure(ex);
            if (IsTransportFailure(ex)) throw new System.IO.IOException("HistorianGateway read transport is unavailable.", ex);
            throw;
        }
    }

    /// <summary>The previous AtTime: AVEVA's own interpolation, aligned to the request (kill switch only).</summary>
    private async Task<HistoryReadResult> ReadAtTimeNativeAsync(
        string fullReference, IReadOnlyList<DateTime> timestampsUtc, CancellationToken cancellationToken)
    {
        try
        {
            if (timestampsUtc.Count > 0 && !await _tags.ExistsAsync(fullReference, cancellationToken).ConfigureAwait(false))
            {
                RecordOutcome(success: true, error: null);
                return MissingAtTime(timestampsUtc, _time.GetUtcNow().UtcDateTime);
            }

            var samples = await _client
                .ReadAtTimeAsync(fullReference, timestampsUtc, cancellationToken)
                .ConfigureAwait(false);
            var aligned = AlignAtTimeSnapshots(timestampsUtc, samples);
            RecordOutcome(success: true, error: null);
            return new HistoryReadResult(aligned, ContinuationPoint: null);
        }
        catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
        {
            throw; // caller-requested, not a backend fault
        }
        catch (Exception ex)
        {
            RecordReadFailure(ex);
            if (IsTransportFailure(ex)) throw new System.IO.IOException("HistorianGateway read transport is unavailable.", ex);
            throw;
        }
    }

    /// <inheritdoc />
    // false: the gateway's ReadRaw streams ascending only and rejects end < start
    // (InvalidArgument). The server serves reverse and end-only capped raw reads from bounded forward reads
    // instead, and never sends this source a reverse window (ReadRawAsync refuses one unsent).
    public bool SupportsReverseRaw => false;

    /// <inheritdoc />
    // The server raw cap (ServerHistorian:MaxRawValuesPerRead; 0 = none), bounded further by a strict gateway's
    // (AllowStreamingBeyondCeiling=false) learned ceiling once a refusal has named it (GatewayRowCeiling) — that
    // ceiling less the RawAggregateReader.PreWindowAllowance a capped read adds for AVEVA's boundary row, so a page of
    // this size asks for exactly the ceiling. The learned part is forgotten when a re-probe finds the gateway streaming
    // again; the cap stays. 0 only with no cap and no learned ceiling.
    public int MaxRawValuesPerRead
    {
        get
        {
            var learned = _ceiling.Learned is var c and > 0 ? Math.Max(1, c - RawAggregateReader.PreWindowAllowance) : 0;
            var cap = _maxRawValuesPerRead;
            return cap > 0 && learned > 0 ? Math.Min(cap, learned) : cap > 0 ? cap : learned;
        }
    }

    /// <inheritdoc />
    // false: the gateway's ReadEvents streams ascending only. The server never sends it a
    // reverse read; it walks one back from bounded forward reads (ties sorted by EventId — the gateway
    // orders by event time only).
    public bool SupportsReverseEvents => false;

    /// <inheritdoc />
    // ServedAggregates with EnableNativeTimeWeighted, else RawEvaluatedAggregates; every other member answers
    // BadAggregateNotSupported and is not advertised.
    public IReadOnlySet<HistoryAggregateType> SupportedAggregates =>
        _enableNativeTimeWeighted ? ServedAggregates : RawEvaluatedAggregates;

    /// <inheritdoc />
    // Data + events, but NOT ServerTimestamps: AVEVA keeps no per-sample server time, so SampleMapper stamps the read
    // time — advertising ServerTimestampSupported would promise a stored value we do not have (lmxopcua#881).
    public HistorianReadCapabilities ReadCapabilities => HistorianReadCapabilities.Data | HistorianReadCapabilities.Events;

    /// <inheritdoc />
    // A reverse (newest-first) read is refused with NotSupportedException — the gateway's
    // ReadEvents streams ascending only and rejects end < start. The server never calls this with
    // reverse: it walks a reverse read back from bounded forward reads. Nothing is sent.
    public Task<HistoricalEventsResult> ReadEventsAsync(
        string? sourceName, DateTime startUtc, DateTime endUtc, int maxEvents, bool reverse,
        CancellationToken cancellationToken) =>
        reverse
            ? throw new NotSupportedException(
                "The HistorianGateway backend reads events oldest-first only; a reverse (newest-first) events read is not supported.")
            : ReadEventsAsync(sourceName, startUtc, endUtc, maxEvents, cancellationToken);

    /// <summary>
    ///     The largest page cap sent as the server-side <c>max_events</c> (as cap + 1): the gateway's default
    ///     <c>Historian:ReadLimits:MaxEventsCeiling</c> is 1,000,000 and a larger <c>max_events</c> is <c>InvalidArgument</c>.
    ///     A larger cap (an uncapped read arrives as <see cref="int.MaxValue"/>) is read without a server cap.
    /// </summary>
    internal const int MaxServerCappedEvents = 999_999;

    /// <inheritdoc />
    // A capped read asks the gateway for cap + 1 events through ReadEventsRequest.max_events (HistorianGateway
    // Contracts/Client 0.4.0+, lmxopcua#746): the gateway stops there and COMPLETES the stream normally — nothing is cancelled,
    // so it never evicts its pooled historian session for this read — and the (cap + 1)-th matching event is the "more
    // events exist" signal. When that answer is ambiguous — the server stopped at cap + 1 rows but the defensive client-side
    // source filter dropped some of them, so whether more matching events exist is unknown — or a gateway refuses the cap
    // (InvalidArgument: a MaxEventsCeiling below cap + 1), the read is repeated without a server cap by the drain below.
    //
    // Uncapped (no server cap) the stream is DRAINED past the page cap — counted, never collected — rather than broken out
    // of: abandoning a gateway stream cancels the RPC, which the gateway classifies as a connection fault and answers by
    // evicting its pooled historian session. The drain is bounded by the EventDrainBudget
    // (ServerHistorian:Gateway:EventDrainBudget): past cap + budget events the stream is abandoned after all, logged at Debug
    // and counted in EventStreamsAbandoned. 0 = abandon at the cap.
    public async Task<HistoricalEventsResult> ReadEventsAsync(
        string? sourceName, DateTime startUtc, DateTime endUtc, int maxEvents,
        CancellationToken cancellationToken)
    {
        try
        {
            var serverCap = maxEvents is > 0 and <= MaxServerCappedEvents ? maxEvents + 1 : 0;
            EventDrain drain;
            if (serverCap > 0)
            {
                try
                {
                    drain = await DrainEventsAsync(sourceName, startUtc, endUtc, maxEvents, serverCap, cancellationToken)
                        .ConfigureAwait(false);
                }
                catch (Exception ex) when (!cancellationToken.IsCancellationRequested && IsInvalidArgument(ex))
                {
                    _logger.LogDebug("HistorianGateway refused the events read's max_events {Cap}; re-reading it without a server cap.",
                        serverCap);
                    drain = await DrainEventsAsync(sourceName, startUtc, endUtc, maxEvents, 0, cancellationToken)
                        .ConfigureAwait(false);
                }
                // Exactly cap + 1 rows and no signal: the server cut the stream, but foreign rows were filtered out of it.
                // (More rows than asked = a gateway that ignores max_events streamed everything: the drain is the answer.)
                if (drain.Received == serverCap && !drain.Truncated)
                {
                    Interlocked.Increment(ref _eventServerCapReReads);
                    drain = await DrainEventsAsync(sourceName, startUtc, endUtc, maxEvents, 0, cancellationToken)
                        .ConfigureAwait(false);
                }
            }
            else
            {
                drain = await DrainEventsAsync(sourceName, startUtc, endUtc, maxEvents, 0, cancellationToken)
                    .ConfigureAwait(false);
            }

            RecordOutcome(success: true, error: null);
            // A non-null, opaque token signals truncation to the caller.
            // The gateway has no resumable cursor, so the token's contents carry no paging state —
            // its presence alone is the "more events exist" signal. A fresh array per call keeps it
            // from being shared/mutated.
            return new HistoricalEventsResult(drain.Collected, drain.Truncated ? new byte[] { 0x01 } : null);
        }
        catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
        {
            throw; // the caller (client cancel / HistoryReadDeadline) asked for this — not a backend fault
        }
        catch (Exception ex)
        {
            RecordReadFailure(ex);
            if (IsTransportFailure(ex)) throw new System.IO.IOException("HistorianGateway read transport is unavailable.", ex);
            throw;
        }
    }

    private long _eventServerCapReReads;

    /// <summary>
    ///     Server-capped events reads repeated without the cap because the capped answer was ambiguous (foreign rows filtered
    ///     out of a stream the server cut at cap + 1) — test/diagnostic.
    /// </summary>
    internal long EventServerCapReReads => Interlocked.Read(ref _eventServerCapReReads);

    /// <summary>One events stream's outcome: the collected page, the "more exist" signal and the stream rows received.</summary>
    private readonly record struct EventDrain(List<HistoricalEvent> Collected, bool Truncated, long Received);

    /// <summary>
    ///     Reads one events stream: collects up to <paramref name="maxEvents"/> matching events, then (cap reached) looks for
    ///     one more matching event as the "more exist" signal and drains the rest within the EventDrainBudget.
    ///     <paramref name="serverCap"/> &gt; 0 is sent as <c>max_events</c>; 0 = no server cap.
    /// </summary>
    private async Task<EventDrain> DrainEventsAsync(string? sourceName, DateTime startUtc, DateTime endUtc, int maxEvents,
        int serverCap, CancellationToken cancellationToken)
    {
        var hasCap = maxEvents > 0;
        // Bounded initial capacity: an uncapped read arrives as int.MaxValue ("Array dimensions exceeded").
        var collected = new List<HistoricalEvent>(hasCap ? Math.Min(maxEvents, 1024) : 0);
        var truncated = false;
        long pastCap = 0; // stream rows consumed after the cap filled (matching or not)
        long received = 0;
        var abandoned = false;

        await foreach (var wireEvent in _client
            .ReadEventsAsync(sourceName, startUtc, endUtc, serverCap, cancellationToken)
            .ConfigureAwait(false))
        {
            received++;
            if (hasCap && collected.Count == maxEvents)
            {
                // The page is full: drain the rest of the stream without collecting it, within the budget.
                // One more MATCHING event is the "more events exist" signal; a foreign source's row is not, and
                // until the signal is seen no row counts toward the budget — abandoning before it would answer
                // "complete" for a page that may not be (and a spurious continuation point on a stretch of foreign
                // rows would make the server crawl it one resolution step per page).
                if (!truncated)
                {
                    if (!EventSourceMatch.Matches(sourceName, wireEvent.SourceName))
                        continue;
                    truncated = true;
                }
                // Server-capped, the stream is already bounded at cap + 1 (by the gateway, or by the package client against an
                // older gateway): read it to its end so the call completes normally instead of being cancelled. Only a stream
                // running past the max_events it was sent is drained within the budget.
                if (serverCap > 0 && received <= serverCap) continue;
                if (++pastCap > _eventDrainBudget)
                {
                    abandoned = true;
                    break;
                }
                continue;
            }

            var mapped = EventMapper.ToHistoricalEvent(wireEvent);

            // Defensive client-side source filter: the gateway's SQL ReadEvents source filter
            // may not be present, so drop any event whose source does not match the request. The request's own
            // events and its attributes' (<source>.<attribute>, the name the gateway reports an AVEVA attribute
            // alarm under) match; a sibling that merely shares the prefix does not (#882, EventSourceMatch).
            if (!EventSourceMatch.Matches(sourceName, mapped.SourceName))
            {
                continue;
            }

            collected.Add(mapped);
        }

        if (abandoned)
        {
            Interlocked.Increment(ref _eventStreamsAbandoned);
            _logger.LogDebug(
                "HistorianGateway events read abandoned its stream {Budget} rows past the page cap (EventDrainBudget); " +
                "the gateway may evict its historian session.", _eventDrainBudget);
        }

        return new EventDrain(collected, truncated, received);
    }

    /// <summary>Whether a gateway failure is an <c>InvalidArgument</c> refusal (raw, or wrapped by the package client).</summary>
    private static bool IsInvalidArgument(Exception ex)
    {
        for (Exception? e = ex; e is not null; e = e.InnerException)
            if (e is RpcException { StatusCode: Grpc.Core.StatusCode.InvalidArgument }) return true;
        return false;
    }

    /// <inheritdoc />
    public HistorianHealthSnapshot GetHealthSnapshot()
    {
        lock (_healthLock)
        {
            return new HistorianHealthSnapshot(
                TotalQueries: _totalQueries,
                TotalSuccesses: _totalSuccesses,
                TotalFailures: _totalFailures,
                ConsecutiveFailures: _consecutiveFailures,
                LastSuccessTime: _lastSuccessUtc,
                LastFailureTime: _lastFailureUtc,
                LastError: _lastError,
                // "Open" is the latest connectivity observation: a read outcome, or a ProbeAsync result from the probe
                // loop (ServerHistorian:Gateway:ProbeInterval), whichever came last — so a node that has served no
                // HistoryRead yet (or only writes) still reports a healthy gateway. Without a probe loop it is read
                // outcomes alone: at least one success and no failure since. The gateway is non-clustered to us, so
                // node fields are null/empty.
                ProcessConnectionOpen: _lastObservationOpen ?? false,
                EventConnectionOpen: _lastObservationOpen ?? false,
                ActiveProcessNode: null,
                ActiveEventNode: null,
                Nodes: []);
        }
    }

    /// <summary>
    ///     Reconciles a gateway at-time reply against the requested timestamps to honour the
    ///     <see cref="IHistorianDataSource.ReadAtTimeAsync(string, IReadOnlyList{DateTime}, CancellationToken)"/> contract: exactly one snapshot per
    ///     requested timestamp, in request order. Returned samples are indexed by timestamp ticks;
    ///     any requested timestamp the gateway did not return is filled with a Bad-quality
    ///     (<c>0x80000000</c>) snapshot stamped at the requested time rather than positionally
    ///     misaligning values. The alignment logic was ported from the now-retired Wonderware
    ///     client's at-time snapshot reconciliation.
    /// </summary>
    private static IReadOnlyList<DataValueSnapshot> AlignAtTimeSnapshots(
        IReadOnlyList<DateTime> timestampsUtc, IReadOnlyList<HistorianSample> samples)
    {
        // Index returned samples by timestamp ticks. Duplicate timestamps keep the first.
        var byTicks = new Dictionary<long, HistorianSample>(samples.Count);
        foreach (var sample in samples)
        {
            if (sample.Timestamp is null) continue;
            byTicks.TryAdd(sample.Timestamp.ToDateTime().Ticks, sample);
        }

        var result = new DataValueSnapshot[timestampsUtc.Count];
        for (var i = 0; i < timestampsUtc.Count; i++)
        {
            var requested = DateTime.SpecifyKind(timestampsUtc[i], DateTimeKind.Utc);
            if (byTicks.TryGetValue(requested.Ticks, out var sample))
            {
                // Reuse the shared sample mapper for value + quality, then re-stamp the source
                // timestamp to the requested time per the ReadAtTime contract.
                result[i] = SampleMapper.ToSnapshot(sample) with { SourceTimestampUtc = requested };
            }
            else
            {
                // Gap — gateway returned no sample for this timestamp. Per the contract this is a
                // Bad-quality snapshot stamped at the requested time, not a dropped row.
                result[i] = new DataValueSnapshot(
                    Value: null,
                    StatusCode: 0x80000000u, // Bad
                    SourceTimestampUtc: requested,
                    ServerTimestampUtc: DateTime.UtcNow);
            }
        }

        return result;
    }

    /// <summary>No-data values for a missing tag, stamped with the client's requested times.</summary>
    private static HistoryReadResult MissingAtTime(IReadOnlyList<DateTime> times, DateTime now) =>
        new(times.Select(t => AtTimeEvaluator.Evaluate(t, null, null, now, now)).ToArray(), ContinuationPoint: null);

    /// <summary>Classifies actual gRPC/package transport failures without treating refusals as outages.</summary>
    private static bool IsTransportFailure(Exception ex) => ex switch
    {
        ZB.MOM.WW.HistorianGateway.Client.HistorianGatewayUnavailableException => true,
        RpcException rpc => rpc.StatusCode is Grpc.Core.StatusCode.Unavailable or Grpc.Core.StatusCode.DeadlineExceeded,
        TimeoutException => true,
        _ => ex.InnerException is { } inner && IsTransportFailure(inner),
    };

    /// <summary>
    ///     Records a failed read: bumps the health counters and logs a generic, redaction-safe
    ///     debug line (no tag, host, or value). The exception itself is rethrown by the caller.
    /// </summary>
    private void RecordReadFailure(Exception ex)
    {
        RecordOutcome(success: false, error: ex.Message);
        _logger.LogDebug("Historian gateway read operation failed and was recorded as a health failure.");
    }

    /// <summary>
    ///     Records the outcome of a single read — increments <c>_totalQueries</c> and exactly one of
    ///     <c>_totalSuccesses</c> / <c>_totalFailures</c> under a single <c>_healthLock</c>
    ///     acquisition so a concurrent <see cref="GetHealthSnapshot"/> never observes a torn state.
    /// </summary>
    private void RecordOutcome(bool success, string? error)
    {
        lock (_healthLock)
        {
            _totalQueries++;
            _lastObservationOpen = success;
            if (success)
            {
                _totalSuccesses++;
                _consecutiveFailures = 0;
                _lastSuccessUtc = DateTime.UtcNow;
            }
            else
            {
                _totalFailures++;
                _consecutiveFailures++;
                _lastFailureUtc = DateTime.UtcNow;
                _lastError = error;
            }
        }
    }

    /// <summary>Disposes the underlying gateway client. Prefer <see cref="DisposeAsync"/>.</summary>
    public void Dispose() => DisposeAsync().AsTask().GetAwaiter().GetResult();

    /// <summary>Stops the probe loop (bounded wait), then asynchronously disposes the underlying gateway client.</summary>
    /// <returns>A task that completes when the client has been disposed.</returns>
    public async ValueTask DisposeAsync()
    {
        if (!_probeStop.IsCancellationRequested)
        {
            _probeStop.Cancel();
            // The loop never throws; a probe stuck past its own bound is abandoned rather than awaited further.
            await Task.WhenAny(_probeLoop, Task.Delay(ProbeCallBound)).ConfigureAwait(false);
            await _client.DisposeAsync().ConfigureAwait(false);
        }
    }
}