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;
namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway;
///
/// Server-side backed by the HistorianGateway gRPC surface
/// (via the seam). Translates OPC UA HistoryRead requests
/// to gateway read calls and maps the wire shapes back to the driver-agnostic
/// / carriers using the pure
/// mappers in Mapping/.
///
///
/// 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-_healthLock discipline ported
/// from WonderwareHistorianClient so TotalSuccesses + TotalFailures == TotalQueries
/// holds at every observed snapshot.
///
public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDisposable
{
///
/// The processed (aggregate) functions the Gateway backend serves — the single source of truth for this backend's
/// aggregate set (#681): the per-sample aggregates evaluated from raw reads (every member of the
/// / /
/// families of : Average, Count, Minimum,
/// Maximum, Range, Start, End, Delta, MinimumActualTime, MaximumActualTime) plus the two answered natively by AVEVA
/// (TimeAverage, Total — ). Any other aggregate throws
/// without calling the gateway. It is what
/// returns, so the advertised AggregateFunctions folders and the server's
/// up-front BadAggregateNotSupported read it too (#682).
///
public static IReadOnlySet ServedAggregates { get; } =
Enum.GetValues()
.Where(a => RawAggregateReader.Serves(a) || AggregateModeMapper.NativeAggregates.Contains(a))
.ToFrozenSet();
private readonly IHistorianGatewayClient _client;
private readonly ILogger _logger;
private readonly int _maxRawRows;
private readonly TimeProvider _time;
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;
/// Creates a gateway-backed historian data source.
/// The gateway client seam used for all reads.
/// Diagnostic logger; failures are recorded without leaking tag/host detail.
///
/// In-window raw-row cap of one raw-evaluated processed read (ServerHistorian:MaxTimeWeightedRawRows;
/// <= 0 = unbounded). Past it the node fails BadTooManyOperations.
///
/// Clock for result server timestamps and the raw look-ahead horizon; null = system.
public GatewayHistorianDataSource(IHistorianGatewayClient client, ILogger logger,
int maxRawRows = TimeWeightedAggregator.DefaultMaxWindowRows, TimeProvider? timeProvider = null)
{
ArgumentNullException.ThrowIfNull(client);
ArgumentNullException.ThrowIfNull(logger);
_client = client;
_logger = logger;
_maxRawRows = maxRawRows;
_time = timeProvider ?? TimeProvider.System;
}
///
///
/// A reverse window (start > end) is refused with before anything is sent
/// and is not recorded as a health failure (Gitea #686): the gateway rejects it InvalidArgument, and the server
/// never asks — is false, so it walks a reverse read back with forward reads.
/// Stored samples only (#687 — Part 11 Raw): AVEVA's synthetic start-boundary row is never returned, see
/// .
///
public async Task 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
{
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 (#687: the pre-window margin swallowed the row budget), not a backend fault.
throw;
}
catch (Exception ex)
{
RecordReadFailure(ex);
throw;
}
}
///
/// The stored samples of [startUtc, endUtc) (a point read start == end: of that instant), oldest first,
/// at most (0 = all) — without AVEVA's synthetic rows (#687).
///
///
///
/// An AVEVA Full raw read starts with a row stamped exactly at the query start: the value in effect there
/// (Quality = 133, 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 ReturnBounds = false wants stored values only, and the
/// server's paging (#669) 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 (#681): it is requested from
/// (1 ms early) and only rows
/// 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.
///
///
/// Row budget. 0 ("all values") asks for (the client package and the
/// gateway reject max_values <= 0, #686). A cap asks for cap +
/// rows — NOT clamped at the gateway's default ceiling (#688
/// 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 refuses an over-ceiling
/// budget instead (⇒ , BadTooManyOperations).
///
///
/// Completeness. 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
/// start − 1 ms) used up the budget, the window is read ONCE more asking for cap + m +
/// (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
/// . Every stream is consumed to completion (never broken out of —
/// a cancelled gateway stream evicts its pooled historian session). ReturnBounds is not plumbed to the
/// data source on any backend, so bounding values are never synthesised here.
///
///
private async Task> 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);
if (cap == 0)
return (await ReadRawPassAsync(fullReference, startUtc, endUtc, cap, int.MaxValue, cancellationToken)
.ConfigureAwait(false)).Samples;
var first = await ReadRawPassAsync(fullReference, startUtc, endUtc, cap,
(int)Math.Min(int.MaxValue, (long)cap + RawAggregateReader.PreWindowAllowance), cancellationToken).ConfigureAwait(false);
if (first.Complete) return first.Samples;
// The margin crowded the window out: one bounded re-read with room for the pre-window rows already seen.
var second = await ReadRawPassAsync(fullReference, startUtc, endUtc, cap,
(int)Math.Min(int.MaxValue, (long)cap + first.PreWindowRows + RawAggregateReader.PreWindowAllowance),
cancellationToken).ConfigureAwait(false);
if (second.Complete) return second.Samples;
throw new HistoryReadLimitExceededException(cap,
$"The raw read of '{fullReference}' could not be read completely: more than {second.PreWindowRows} rows lie in " +
$"the {RawAggregateReader.BoundaryMargin.TotalMilliseconds} ms before the window and exhausted the row budget.");
}
/// One forward raw read of a window, drained, with its pre-window row count and whether it is complete.
private async Task<(List Samples, int PreWindowRows, bool Complete)> ReadRawPassAsync(
string fullReference, DateTime startUtc, DateTime endUtc, int cap, int budget, CancellationToken cancellationToken)
{
var samples = new List();
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);
}
else if (sample.Timestamp is { } ts && ts.ToDateTime() < startUtc) preWindow++;
else if (sample.Timestamp is not null) reachedEnd = true;
}
var complete = received < budget || samples.Count >= cap || reachedEnd || cap == 0;
return (samples, preWindow, complete);
}
///
///
///
/// Per-sample aggregates (#681) — Average, Count, Minimum, Maximum, Range, Start, End, Delta,
/// Minimum/MaximumActualTime — are evaluated from a raw read with the SQL backends' shared evaluator
/// (): identical values, status, Calculated/Partial bits and timestamps to
/// SqlServer/QuestDB for the same samples, forward and reverse (#677). The in-window raw rows are capped by
/// ServerHistorian:MaxTimeWeightedRawRows (, not counted as
/// a backend failure).
///
///
/// TimeAverage / Total stay native AVEVA retrieval modes (), 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 (#685, ). Everything else is
/// (see ).
///
///
/// A raw-evaluated read whose window holds more than
/// intervals.
public async Task 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 (#285).
var mode = AggregateModeMapper.ToRetrievalMode(aggregate);
try
{
// TimeAverage / Total are native gateway retrieval modes (TimeWeightedAverage / Integral) — no
// client-side scaling.
var buckets = new List();
await foreach (var bucket in _client
.ReadAggregateAsync(fullReference, startUtc, endUtc, mode, interval, cancellationToken)
.ConfigureAwait(false))
{
buckets.Add(bucket);
}
// Filtered while mapping, after the stream was drained (#685: AVEVA's extra pre-window row).
var snapshots = SampleMapper.ToAggregateSnapshots(buckets, startUtc, endUtc);
RecordOutcome(success: true, error: null);
return new HistoryReadResult(snapshots, ContinuationPoint: null);
}
catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
{
throw; // caller-requested, not a backend fault
}
catch (Exception ex)
{
RecordReadFailure(ex);
throw;
}
}
/// Serves a per-sample aggregate from a raw read (#681); see .
private async Task 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;
var values = await RawAggregateReader
.ReadAsync(_client, 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);
throw;
}
}
///
/// True when (or an inner exception — the client package wraps the )
/// is the gateway's InvalidArgument "max_values exceeds the server ceiling" rejection.
///
internal static bool IsMaxValuesCeilingRejection(Exception ex)
{
for (var e = ex; e is not null; e = e.InnerException)
if (e is RpcException { StatusCode: StatusCode.InvalidArgument } rpc
&& rpc.Status.Detail.Contains("max_values exceeds the server ceiling", StringComparison.Ordinal))
return true;
return false;
}
///
public async Task ReadAtTimeAsync(
string fullReference, IReadOnlyList timestampsUtc, CancellationToken cancellationToken)
{
try
{
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 (Exception ex)
{
RecordReadFailure(ex);
throw;
}
}
///
///
/// false (Gitea #686): 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 ( refuses one unsent).
///
public bool SupportsReverseRaw => false;
///
///
/// false: the gateway's ReadEvents streams ascending only (Gitea #675). The server therefore refuses an
/// explicit reverse events read without calling, and keeps an end-only capped read forward.
///
public bool SupportsReverseEvents => false;
///
/// ; every other member answers BadAggregateNotSupported and is not advertised.
public IReadOnlySet SupportedAggregates => ServedAggregates;
///
///
/// A reverse (newest-first) read is refused with — the server reports
/// BadHistoryOperationUnsupported (Gitea #675). The gateway's ReadEvents streams ascending only and
/// rejects end < start, so the only way to serve "the latest N" would be to drain the whole window and
/// keep its tail, an unbounded read the cap exists to prevent. Nothing is sent.
///
public Task 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);
///
public async Task ReadEventsAsync(
string? sourceName, DateTime startUtc, DateTime endUtc, int maxEvents,
CancellationToken cancellationToken)
{
try
{
var hasCap = maxEvents > 0;
var collected = new List(hasCap ? maxEvents : 0);
var truncated = false;
await foreach (var wireEvent in _client
.ReadEventsAsync(sourceName, startUtc, endUtc, maxEvents, cancellationToken)
.ConfigureAwait(false))
{
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.
if (sourceName is not null && !string.Equals(mapped.SourceName, sourceName, StringComparison.Ordinal))
{
continue;
}
// One more matching event arriving once the cap is full means the result is
// truncated — stop draining and flag it.
if (hasCap && collected.Count == maxEvents)
{
truncated = true;
break;
}
collected.Add(mapped);
}
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(collected, truncated ? new byte[] { 0x01 } : null);
}
catch (Exception ex)
{
RecordReadFailure(ex);
throw;
}
}
///
public HistorianHealthSnapshot GetHealthSnapshot()
{
lock (_healthLock)
{
return new HistorianHealthSnapshot(
TotalQueries: _totalQueries,
TotalSuccesses: _totalSuccesses,
TotalFailures: _totalFailures,
ConsecutiveFailures: _consecutiveFailures,
LastSuccessTime: _lastSuccessUtc,
LastFailureTime: _lastFailureUtc,
LastError: _lastError,
// No connection probe exists (the old RefreshConnectionStateAsync had no production
// caller and was deleted, archreview 06/S-6): "open" is derived from observed reads
// alone -- at least one success and no failure since. The gateway is non-clustered to
// us, so node fields are null/empty (mirrors the Wonderware client's Finding 010 posture).
ProcessConnectionOpen: _totalSuccesses > 0 && _consecutiveFailures == 0,
EventConnectionOpen: _totalSuccesses > 0 && _consecutiveFailures == 0,
ActiveProcessNode: null,
ActiveEventNode: null,
Nodes: []);
}
}
///
/// Reconciles a gateway at-time reply against the requested timestamps to honour the
/// 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
/// (0x80000000) 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.
///
private static IReadOnlyList AlignAtTimeSnapshots(
IReadOnlyList timestampsUtc, IReadOnlyList samples)
{
// Index returned samples by timestamp ticks. Duplicate timestamps keep the first.
var byTicks = new Dictionary(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;
}
///
/// 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.
///
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.");
}
///
/// Records the outcome of a single read — increments _totalQueries and exactly one of
/// _totalSuccesses / _totalFailures under a single _healthLock
/// acquisition so a concurrent never observes a torn state.
///
private void RecordOutcome(bool success, string? error)
{
lock (_healthLock)
{
_totalQueries++;
if (success)
{
_totalSuccesses++;
_consecutiveFailures = 0;
_lastSuccessUtc = DateTime.UtcNow;
}
else
{
_totalFailures++;
_consecutiveFailures++;
_lastFailureUtc = DateTime.UtcNow;
_lastError = error;
}
}
}
/// Disposes the underlying gateway client. Prefer .
public void Dispose() => DisposeAsync().AsTask().GetAwaiter().GetResult();
/// Asynchronously disposes the underlying gateway client.
/// A task that completes when the client has been disposed.
public ValueTask DisposeAsync() => _client.DisposeAsync();
}