fix(historian): fail fast for missing tags and reduce gateway read cost
This commit is contained in:
+16
-2
@@ -765,8 +765,11 @@ data.
|
||||
refusal of the remembered one. A **raised** ceiling (or a switch to streaming) is noticed by the only reads it matters to: when a
|
||||
read bounded by the remembered ceiling comes back full without proving completeness, it is re-probed **once** at its full budget
|
||||
— served ⇒ the ceiling is forgotten; refused naming a larger ceiling ⇒ that one is learned and the read re-issued at it; refused
|
||||
naming the same ceiling ⇒ the bounded result stands. No timer: a TTL would only add refusals for reads that already fit; the cost
|
||||
of the rule is one refused re-probe per genuinely over-ceiling read (which fails either way). So an uncapped read (`NumValuesPerNode = 0`, which used to be refused outright on a strict gateway, even for a 10-row
|
||||
naming the same ceiling ⇒ the bounded result stands. Raw and AtTime re-probe immediately. Per-sample aggregates (#696) reuse
|
||||
a refusal for **30 seconds** before re-probing a full stream; within that period an oversized window still fails honestly with
|
||||
`BadTooManyOperations`. A lowered ceiling is still re-learned immediately; a raise becomes visible on the next incomplete
|
||||
aggregate read after the cooldown. So an uncapped read (`NumValuesPerNode = 0`, which used to be refused outright on a strict
|
||||
gateway, even for a 10-row
|
||||
window) and a page size within 16 of the ceiling are now served when the window fits. A ceiling-bounded read is answered only
|
||||
when it is **provably complete** — the stream came back short of the ceiling, or reached `EndTime` (or, capped, filled the page);
|
||||
a stream that delivered the whole ceiling short of `EndTime` may be truncated and fails **`BadTooManyOperations`** (not a backend
|
||||
@@ -828,6 +831,17 @@ data.
|
||||
refusing; doubling the look-back allowance does not skip that final usable budget. Reverse paging retries a crowded segment
|
||||
with a smaller in-window budget derived from the learned ceiling and the observed pre-window rows.
|
||||
|
||||
**Missing tags (#653).** Gateway Raw, Processed and AtTime first check the tag catalog. An empty exact-name catalog result or an explicit
|
||||
`NotFound` returns no data without a history RPC: Raw is empty; AtTime and Processed return null/`BadNoData` at the
|
||||
requested times or bucket timestamps. Unsupported aggregate validation still runs first. A bounded per-data-source cache
|
||||
holds at most 4,096 tag names, with **5 seconds** for absence and **60 seconds** for presence, so newly provisioned tags become
|
||||
visible after the short negative expiry. Catalog faults and cancellation never become missing-tag cache entries. The adapter drains the filtered catalog stream and checks exact names, avoiding older gateways that misparse a missing `GetTagMetadata` response as `Internal`. Empty
|
||||
AtTime/Processed requests need no check. Events and modified-history capability checks are unaffected.
|
||||
|
||||
**Stopped-tag reverse discovery (#696).** On a strict Gateway, a truncated remainder probe retains its oldest-row evidence
|
||||
and density. Gap discovery uses the learned ceiling's available budget; ordinary page reads keep their page-sized budget.
|
||||
Only complete segments contribute to a page, preserving newest-first order, bounds, ties and continuation hints.
|
||||
|
||||
### Gateway AtTime is evaluated from raw reads (#693)
|
||||
|
||||
> ⚠️ **Client-visible (#693).** Gateway AtTime used to pass AVEVA's own interpolated answer (the gateway's `ReadAtTime`) through,
|
||||
|
||||
@@ -0,0 +1,36 @@
|
||||
# Missing-tag and Gateway performance fixes — 2026-10-13
|
||||
|
||||
Follow-up to pushed `97668d26` / PR #699. Implements #653 and the remaining #696 items 2/3 under the approved bounded design.
|
||||
|
||||
## Changes
|
||||
|
||||
- **#653:** Raw, Processed and AtTime check tag presence before history RPCs. The bounded 4,096-name cache uses a 5-second negative and 60-second positive lifetime. Only explicit absence or NotFound is cached as missing; failures and cancellation propagate. Raw is empty; AtTime and Processed retain requested/bucket timestamps with null/BadNoData. Unsupported aggregate validation runs first.
|
||||
- **Live compatibility adjustment:** the deployed gateway misparses missing `GetTagMetadata` replies as `Internal` (`InvalidDataException: GetTagInfo response ended unexpectedly`). The adapter therefore checks the filtered tag catalog and compares exact names, fully draining the stream before declaring presence/absence. It never turns the parser error into a missing tag. Wildcard matches alone do not prove presence, and a trailing stream fault cannot enter the cache.
|
||||
- **#696 item 2:** a truncated remainder probe seeds strict-ceiling reverse discovery with its observed rows/density; gap searches use the available discovery budget. Complete segments alone contribute to the newest-first page. Regression: a 7,200-row stopped 1 Hz tag, ceiling 1,000, requested latest three one day later, previously 17 raw calls; now meets a maximum of 12 with exactly 7199/7198/7197 returned. Ordinary page sizing and event discovery behavior are unchanged.
|
||||
- **#696 item 3:** per-sample aggregates reuse a ceiling refusal for 30 seconds instead of immediately re-probing on every full stream. The next incomplete aggregate read after expiry notices a raised ceiling; a lowered ceiling is re-learned immediately. Raw and AtTime re-probes remain immediate. During the cooldown an oversized aggregate remains BadTooManyOperations, never a truncated result.
|
||||
|
||||
## Verification evidence
|
||||
|
||||
New regressions failed before fixes: all initial 12 cases failed (missing preflight/caching, 17 discovery reads, duplicate aggregate refusal). The four catalog wire cases failed with the metadata adapter and pass with the catalog adapter. Final Gateway suite: 434 passed, six environment-gated skips.
|
||||
|
||||
Independent read-only review found no important defects. Concurrent cold requests can duplicate a metadata/catalog check; cache access and capacity mutations are locked. No single-flight guarantee was introduced. Catalog RPC contract streams all matches; the upstream SDK pages the native tag-query results, with no configured browse result ceiling.
|
||||
|
||||
## Live #653 confirmation
|
||||
|
||||
Measured the updated GatewayHistorianDataSource through the real adapter and local gateway against the existing remote AVEVA connection. Temporary historian:read key had a two-hour expiry and was revoked after the run. No historian writes/provisioning and no gateway deployment changes.
|
||||
|
||||
Nonexistent tag: `OtOpcUa.Issue653.Nonexistent.20261013`.
|
||||
|
||||
| Operation, five consecutive runs | Elapsed milliseconds | Result |
|
||||
|---|---|---|
|
||||
| Raw | 228, 0, 0, 0, 0 | empty |
|
||||
| Count processed | 2, 0, 0, 0, 0 | six null/BadNoData buckets |
|
||||
| AtTime | 1, 0, 0, 0, 0 | null/BadNoData at requested time |
|
||||
|
||||
These used one data source/cache: the first Raw call is cold, subsequent calls are cached. The prior direct-client measurement took ~30.6 seconds for missing-tag Raw/Aggregate and ~61.3 seconds for AtTime. Direct client RPCs are not modified; the fast-fail protection is in OtOpcUa.
|
||||
|
||||
**Control limitation:** SysTimeSec catalog presence succeeded, but its Raw RPC twice timed out upstream with DeadlineExceeded at ~30 seconds. The data source propagated IOException/communication failure rather than returning no data. A fresh control using the original package client (no data-source presence checks) also timed out with DeadlineExceeded after 30,656 ms, confirming this timeout is independent of the new checks. The earlier direct-client control had succeeded. This run therefore verifies missing-tag fast-fail and failure propagation, not healthy-tag live history behavior. No end-to-end deployed OPC UA claim is made.
|
||||
|
||||
#694 remains open for native aggregate rerun after historiangw#27. No sibling gateway code changed.
|
||||
|
||||
Full final-tree solution: 14,600 passed, 609 skipped, 0 failed across 62 suites. `dotnet test ZB.MOM.WW.OtOpcUa.slnx --verbosity quiet`, exit 0. Includes 122 passing OPC UA integration tests and 268 passing host integration tests (12 host skips). Environment-gated suites are included in the skip count.
|
||||
+34
-1
@@ -68,7 +68,8 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis
|
||||
internal int MaxPreWindowRows => _maxPreWindowRows;
|
||||
|
||||
/// <summary>A strict gateway's row ceiling, learned from its first refusal (#691.3).</summary>
|
||||
private readonly GatewayRowCeiling _ceiling = new();
|
||||
private readonly GatewayRowCeiling _ceiling;
|
||||
private readonly GatewayTagPresenceCache _tags;
|
||||
|
||||
private readonly object _healthLock = new();
|
||||
private DateTime? _lastSuccessUtc;
|
||||
@@ -107,6 +108,8 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis
|
||||
_logger = logger;
|
||||
_maxRawRows = maxRawRows;
|
||||
_time = timeProvider ?? TimeProvider.System;
|
||||
_ceiling = new GatewayRowCeiling(_time);
|
||||
_tags = new GatewayTagPresenceCache(_client, _time);
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
@@ -126,6 +129,11 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis
|
||||
|
||||
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);
|
||||
@@ -351,6 +359,11 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis
|
||||
if (plan.BucketCount == 0) return new HistoryReadResult([], ContinuationPoint: null);
|
||||
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 NativeAggregatePlan.Requests(plan))
|
||||
{
|
||||
@@ -390,6 +403,11 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis
|
||||
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);
|
||||
@@ -460,6 +478,11 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis
|
||||
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);
|
||||
@@ -488,6 +511,12 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis
|
||||
{
|
||||
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);
|
||||
@@ -669,6 +698,10 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis
|
||||
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 (#697).</summary>
|
||||
private static bool IsTransportFailure(Exception ex) => ex switch
|
||||
{
|
||||
|
||||
@@ -76,13 +76,15 @@ internal static class GatewayReadLimits
|
||||
/// the read is re-issued at <c>N</c>; refused naming <c>C</c> again ⇒ the <c>C</c>-bounded result stands (no repeat).
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// <b>Why no TTL.</b> A raise only matters to a read that needs more than <c>C</c>, and such a read re-probes anyway
|
||||
/// (rule 3), so a timer would only add refusals for reads that fit. The cost of the rule is one refused re-probe per
|
||||
/// genuinely over-ceiling read — a read that fails <c>BadTooManyOperations</c> whatever is done.
|
||||
/// <b>Re-probe timing (#696).</b> Raw and AtTime reads re-probe incomplete streams immediately. Per-sample
|
||||
/// aggregate reads opt into a 30-second cooldown after a refusal; a full stream during it remains a request
|
||||
/// limit, never a truncated answer. After the cooldown the next incomplete read detects a raised ceiling.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
internal sealed class GatewayRowCeiling
|
||||
internal sealed class GatewayRowCeiling(TimeProvider? timeProvider = null)
|
||||
{
|
||||
private readonly TimeProvider _time = timeProvider ?? TimeProvider.System;
|
||||
private long _lastRefusalTicks;
|
||||
private int _learned; // 0 = unknown / not strict
|
||||
|
||||
/// <summary>The learned ceiling (0 = none).</summary>
|
||||
@@ -93,8 +95,9 @@ internal sealed class GatewayRowCeiling
|
||||
/// <param name="budget">The <c>max_values</c> the read wants.</param>
|
||||
/// <param name="pass">One drained read with the given budget.</param>
|
||||
/// <param name="needsMore">True when a pass with the given budget delivered it all without proving completeness.</param>
|
||||
/// <param name="deferRecentReprobe">Aggregate-only opt-in: reuse a refusal for 30 seconds before re-probing an incomplete read.</param>
|
||||
/// <returns>The result and the budget it was read with.</returns>
|
||||
public async Task<(T Result, int Budget)> ReadAsync<T>(int budget, Func<int, Task<T>> pass, Func<T, int, bool> needsMore)
|
||||
public async Task<(T Result, int Budget)> ReadAsync<T>(int budget, Func<int, Task<T>> pass, Func<T, int, bool> needsMore, bool deferRecentReprobe = false)
|
||||
{
|
||||
var learned = Learned;
|
||||
if (learned <= 0 || learned >= budget) return await AttemptAsync(budget, pass).ConfigureAwait(false);
|
||||
@@ -102,6 +105,10 @@ internal sealed class GatewayRowCeiling
|
||||
var bounded = await AttemptAsync(learned, pass).ConfigureAwait(false);
|
||||
if (bounded.Budget < learned || !needsMore(bounded.Result, bounded.Budget)) return bounded;
|
||||
|
||||
// #696: aggregate reads may reuse a recent refusal for 30 seconds. Raw paging keeps immediate re-probes.
|
||||
if (deferRecentReprobe && _time.GetUtcNow().UtcTicks - Volatile.Read(ref _lastRefusalTicks) < TimeSpan.FromSeconds(30).Ticks)
|
||||
return bounded;
|
||||
|
||||
// Rule 3: full at a possibly stale ceiling — re-probe once at the full budget.
|
||||
int fresh;
|
||||
try
|
||||
@@ -114,10 +121,16 @@ internal sealed class GatewayRowCeiling
|
||||
{
|
||||
// fall through (outside the catch)
|
||||
}
|
||||
Volatile.Write(ref _learned, fresh);
|
||||
Learn(fresh);
|
||||
return fresh > learned ? (await pass(fresh).ConfigureAwait(false), fresh) : bounded;
|
||||
}
|
||||
|
||||
private void Learn(int ceiling)
|
||||
{
|
||||
Volatile.Write(ref _lastRefusalTicks, _time.GetUtcNow().UtcTicks);
|
||||
Volatile.Write(ref _learned, ceiling);
|
||||
}
|
||||
|
||||
private async Task<(T Result, int Budget)> AttemptAsync<T>(int budget, Func<int, Task<T>> pass)
|
||||
{
|
||||
int ceiling;
|
||||
@@ -129,7 +142,7 @@ internal sealed class GatewayRowCeiling
|
||||
{
|
||||
// fall through to the retry (outside the catch, so a second refusal is not nested in this one)
|
||||
}
|
||||
Volatile.Write(ref _learned, ceiling);
|
||||
Learn(ceiling);
|
||||
return (await pass(ceiling).ConfigureAwait(false), ceiling);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,36 @@
|
||||
using Grpc.Core;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway;
|
||||
|
||||
/// <summary>Bounded metadata cache (#653). Missing tags expire quickly; failures never become missing tags.</summary>
|
||||
internal sealed class GatewayTagPresenceCache(IHistorianGatewayClient client, TimeProvider time)
|
||||
{
|
||||
internal const int Capacity = 4096;
|
||||
private readonly object _sync = new();
|
||||
private readonly Dictionary<string, (bool Exists, DateTimeOffset Expires)> _entries = new(StringComparer.Ordinal);
|
||||
|
||||
public async Task<bool> ExistsAsync(string tag, CancellationToken ct)
|
||||
{
|
||||
ct.ThrowIfCancellationRequested();
|
||||
lock (_sync)
|
||||
if (_entries.TryGetValue(tag, out var entry) && time.GetUtcNow() < entry.Expires) return entry.Exists;
|
||||
|
||||
bool exists;
|
||||
try { exists = await client.TagExistsAsync(tag, ct).ConfigureAwait(false); }
|
||||
catch (Exception ex) when (IsNotFound(ex)) { exists = false; }
|
||||
ct.ThrowIfCancellationRequested();
|
||||
lock (_sync)
|
||||
{
|
||||
if (!_entries.ContainsKey(tag) && _entries.Count >= Capacity) _entries.Remove(_entries.Keys.First());
|
||||
_entries[tag] = (exists, time.GetUtcNow().AddSeconds(exists ? 60 : 5));
|
||||
}
|
||||
return exists;
|
||||
}
|
||||
|
||||
private static bool IsNotFound(Exception ex)
|
||||
{
|
||||
for (Exception? e = ex; e is not null; e = e.InnerException)
|
||||
if (e is RpcException { StatusCode: StatusCode.NotFound }) return true;
|
||||
return false;
|
||||
}
|
||||
}
|
||||
+11
@@ -81,6 +81,17 @@ public sealed class HistorianGatewayClientAdapter : IHistorianGatewayClient, IDi
|
||||
return new HistorianGatewayClientAdapter(HistorianGatewayClient.Create(clientOptions));
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public async Task<bool> TagExistsAsync(string tag, CancellationToken ct)
|
||||
{
|
||||
// #653: older gateway SDKs misparse a missing GetTagInfo response as Internal. A catalog query proves
|
||||
// absence without guessing from an error. Even a glob-like tag is checked by exact name, never by row count.
|
||||
var found = false;
|
||||
await foreach (var name in _inner.BrowseTagNamesAsync(tag, ct).ConfigureAwait(false))
|
||||
found |= string.Equals(name, tag, StringComparison.OrdinalIgnoreCase);
|
||||
return found; // drain first: a trailing stream fault must propagate and must not become a cache entry
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public IAsyncEnumerable<HistorianSample> ReadRawAsync(
|
||||
string tag, DateTime startUtc, DateTime endUtc, int maxValues, CancellationToken ct) =>
|
||||
|
||||
@@ -11,6 +11,12 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway;
|
||||
/// </summary>
|
||||
public interface IHistorianGatewayClient : IAsyncDisposable
|
||||
{
|
||||
/// <summary>Checks the tag catalog; false means explicitly absent, while failures propagate.</summary>
|
||||
/// <param name="tag">Historian tag name.</param>
|
||||
/// <param name="ct">Cancellation token.</param>
|
||||
/// <returns>Whether the historian contains the tag.</returns>
|
||||
Task<bool> TagExistsAsync(string tag, CancellationToken ct);
|
||||
|
||||
/// <summary>Streams raw historian samples for a tag over a time window.</summary>
|
||||
/// <param name="tag">The historian tag name to read.</param>
|
||||
/// <param name="startUtc">Inclusive start of the requested time window (UTC).</param>
|
||||
|
||||
@@ -188,7 +188,7 @@ internal static class RawAggregateReader
|
||||
var requested = MaxValuesFor(maxWindowRows);
|
||||
var (pass, maxValues) = await ceiling.ReadAsync(requested,
|
||||
budget => ReadWindowAsync(client, tag, aggregate, plan, requestStart, budget, maxWindowRows, ct),
|
||||
(r, budget) => !r.OverCap && r.Received >= budget).ConfigureAwait(false);
|
||||
(r, budget) => !r.OverCap && r.Received >= budget, deferRecentReprobe: true).ConfigureAwait(false);
|
||||
var (buckets, earliest, latest, received, overCap) = pass;
|
||||
if (overCap) throw LimitExceeded(tag, maxWindowRows);
|
||||
if (received >= maxValues) throw BudgetExhausted(tag, maxWindowRows, requested, maxValues);
|
||||
|
||||
@@ -126,7 +126,7 @@ internal static class ReverseRawReader
|
||||
(budget < Math.Max(1, maxTieCluster)
|
||||
? "the backend's row ceiling (HistorianGateway Historian:ReadLimits:MaxValuesCeiling)."
|
||||
: "ServerHistorian:MaxTieClusterOverfetch.")),
|
||||
ct)
|
||||
ct, useDiscoveryBudgetForGaps: limit > 0)
|
||||
.ConfigureAwait(false);
|
||||
return new ReverseRawPage(rows, next.WidthTicks) { Hint = next };
|
||||
}
|
||||
|
||||
@@ -123,13 +123,14 @@ internal static class ReverseWalk
|
||||
/// <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)
|
||||
ReverseWalkHint hint, Func<DateTime, Exception> tooManyAtOneTimestamp, CancellationToken ct, bool useDiscoveryBudgetForGaps = false)
|
||||
{
|
||||
var resolution = Math.Max(1L, resolutionTicks);
|
||||
var granularity = Math.Max(Alignment, resolution);
|
||||
@@ -206,7 +207,8 @@ internal static class ReverseWalk
|
||||
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 ? discoveryBudget : pageBudget;
|
||||
var segBudget = (discovering && !probe) || (useDiscoveryBudgetForGaps && (probe || known > 0))
|
||||
? discoveryBudget : pageBudget;
|
||||
var cap = segBudget + 2;
|
||||
IReadOnlyList<T> rows;
|
||||
var narrowed = false;
|
||||
@@ -220,10 +222,14 @@ internal static class ReverseWalk
|
||||
|
||||
if (probe)
|
||||
{
|
||||
// Deep history below: the probe saw only its oldest rows. Carry on walking from the same segHi.
|
||||
// #696: strict-ceiling discovery retains the rows/density instead of discarding the remainder probe.
|
||||
probed = true;
|
||||
narrowed = true;
|
||||
break;
|
||||
if (!useDiscoveryBudgetForGaps)
|
||||
{
|
||||
narrowed = true;
|
||||
break;
|
||||
}
|
||||
probe = false;
|
||||
}
|
||||
|
||||
var leading = 0;
|
||||
@@ -284,7 +290,7 @@ internal static class ReverseWalk
|
||||
// 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) * (pageBudget / 2.0) / (inner - 1), maxWidth)
|
||||
? (long)Math.Min((double)(lastInner - firstInner) * ((useDiscoveryBudgetForGaps ? segBudget : pageBudget) / 2.0) / (inner - 1), maxWidth)
|
||||
: resolution;
|
||||
known = lastInner;
|
||||
knownWidth = Math.Max(resolution, aimed);
|
||||
|
||||
+11
@@ -12,6 +12,17 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests;
|
||||
/// </summary>
|
||||
public sealed class FakeHistorianGatewayClient : IHistorianGatewayClient
|
||||
{
|
||||
public bool TagExistsResult = true;
|
||||
public int TagExistsCallCount;
|
||||
public Exception? TagExistsThrows;
|
||||
public Task<bool> TagExistsAsync(string tag, CancellationToken ct)
|
||||
{
|
||||
ct.ThrowIfCancellationRequested();
|
||||
TagExistsCallCount++;
|
||||
if (TagExistsThrows is { } error) throw error;
|
||||
return Task.FromResult(TagExistsResult);
|
||||
}
|
||||
|
||||
// ---- ReadRaw -------------------------------------------------------------------------------
|
||||
public IReadOnlyList<HistorianSample> RawSamples = Array.Empty<HistorianSample>();
|
||||
public string? LastReadRawTag;
|
||||
|
||||
+15
@@ -385,6 +385,21 @@ public sealed class GatewayCeilingPagingTests : IDisposable
|
||||
finally { await host.DisposeAsync(); }
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task A_stopped_dense_tag_under_a_strict_ceiling_is_found_with_bounded_discovery_cost()
|
||||
{
|
||||
var stored = PerSecond(7200);
|
||||
var fake = Gateway(stored, Ceiling);
|
||||
var ds = new GatewayHistorianDataSource(fake, NullLogger<GatewayHistorianDataSource>.Instance);
|
||||
await ds.ReadRawAsync("T", At(0), At(1), 5000, Ct);
|
||||
fake.RawCalls.Clear();
|
||||
var page = await ReverseRawReader.ReadAsync(ds, "T", At(7200).AddDays(1), DateTime.UnixEpoch,
|
||||
3, 65536, At(7200).AddDays(1), 0, Ct);
|
||||
page.Samples.Select(s => s.Value).ShouldBe(new object?[] { 7199d, 7198d, 7197d });
|
||||
fake.RawCalls.Count.ShouldBeLessThanOrEqualTo(12);
|
||||
EveryRequestFits(fake);
|
||||
}
|
||||
|
||||
// ---- F3: rows crowding the look-back before a page start ---------------------------------------------------------------
|
||||
|
||||
[Theory]
|
||||
|
||||
+126
@@ -0,0 +1,126 @@
|
||||
using Google.Protobuf.WellKnownTypes;
|
||||
using Grpc.Core;
|
||||
using Microsoft.Extensions.Logging.Abstractions;
|
||||
using Shouldly;
|
||||
using Xunit;
|
||||
using ZB.MOM.WW.HistorianGateway.Contracts.Grpc;
|
||||
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests;
|
||||
|
||||
public sealed class GatewayMissingTagTests
|
||||
{
|
||||
private static CancellationToken Ct => TestContext.Current.CancellationToken;
|
||||
private static readonly DateTime Now = new(2026, 10, 13, 8, 0, 0, DateTimeKind.Utc);
|
||||
private sealed class Clock : TimeProvider
|
||||
{
|
||||
public DateTime Current = Now;
|
||||
public override DateTimeOffset GetUtcNow() => new(Current);
|
||||
}
|
||||
private static GatewayHistorianDataSource Source(FakeHistorianGatewayClient fake, Clock? clock = null, bool native = false) =>
|
||||
new(fake, NullLogger<GatewayHistorianDataSource>.Instance, timeProvider: clock, atTimeFromRaw: !native);
|
||||
|
||||
[Theory]
|
||||
[InlineData("Raw")]
|
||||
[InlineData("AtTime")]
|
||||
[InlineData("NativeAtTime")]
|
||||
[InlineData("Count")]
|
||||
[InlineData("TimeAverage")]
|
||||
public async Task A_missing_tag_returns_no_data_without_a_history_RPC(string kind)
|
||||
{
|
||||
var fake = new FakeHistorianGatewayClient { TagExistsResult = false };
|
||||
var ds = Source(fake, native: kind == "NativeAtTime");
|
||||
var result = kind switch
|
||||
{
|
||||
"Raw" => await ds.ReadRawAsync("Missing", Now.AddMinutes(-1), Now, 10, Ct),
|
||||
"AtTime" or "NativeAtTime" => await ds.ReadAtTimeAsync("Missing", [Now], Ct),
|
||||
_ => await ds.ReadProcessedAsync("Missing", Now.AddMinutes(-1), Now, TimeSpan.FromMinutes(1),
|
||||
kind == "Count" ? HistoryAggregateType.Count : HistoryAggregateType.TimeAverage, Ct),
|
||||
};
|
||||
if (kind == "Raw") result.Samples.ShouldBeEmpty();
|
||||
else
|
||||
{
|
||||
result.Samples.Count.ShouldBe(1);
|
||||
result.Samples[0].Value.ShouldBeNull();
|
||||
(result.Samples[0].StatusCode & 0xFFFF0000u).ShouldBe(0x809B0000u);
|
||||
result.Samples[0].SourceTimestampUtc.ShouldBe(kind.Contains("AtTime") ? Now : Now.AddMinutes(-1));
|
||||
}
|
||||
fake.TagExistsCallCount.ShouldBe(1);
|
||||
(fake.ReadRawCallCount + fake.ReadAggregateCallCount + fake.ReadAtTimeCallCount).ShouldBe(0);
|
||||
ds.GetHealthSnapshot().TotalFailures.ShouldBe(0);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Negative_cache_expires_so_a_newly_provisioned_tag_can_be_read()
|
||||
{
|
||||
var fake = new FakeHistorianGatewayClient { TagExistsResult = false };
|
||||
var clock = new Clock(); var ds = Source(fake, clock);
|
||||
await ds.ReadRawAsync("T", Now.AddMinutes(-1), Now, 10, Ct);
|
||||
fake.TagExistsResult = true;
|
||||
fake.RawSamples = [new HistorianSample { Timestamp = Timestamp.FromDateTime(Now.AddSeconds(-1)), NumericValue = 42, OpcQuality = 192 }];
|
||||
(await ds.ReadRawAsync("T", Now.AddMinutes(-1), Now, 10, Ct)).Samples.ShouldBeEmpty();
|
||||
fake.TagExistsCallCount.ShouldBe(1);
|
||||
clock.Current = Now.AddSeconds(5);
|
||||
(await ds.ReadRawAsync("T", Now.AddMinutes(-1), Now, 10, Ct)).Samples.Single().Value.ShouldBe(42d);
|
||||
fake.TagExistsCallCount.ShouldBe(2);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Positive_cache_avoids_checks_per_page_and_expires_after_a_minute()
|
||||
{
|
||||
var fake = new FakeHistorianGatewayClient(); var clock = new Clock(); var ds = Source(fake, clock);
|
||||
await ds.ReadRawAsync("T", Now.AddMinutes(-1), Now, 10, Ct);
|
||||
await ds.ReadRawAsync("T", Now.AddMinutes(-1), Now, 10, Ct);
|
||||
fake.TagExistsCallCount.ShouldBe(1);
|
||||
clock.Current = Now.AddSeconds(60);
|
||||
await ds.ReadRawAsync("T", Now.AddMinutes(-1), Now, 10, Ct);
|
||||
fake.TagExistsCallCount.ShouldBe(2);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task An_explicit_nested_NotFound_is_cached_as_missing()
|
||||
{
|
||||
var fake = new FakeHistorianGatewayClient { TagExistsThrows = new InvalidOperationException("wrapper",
|
||||
new RpcException(new Status(StatusCode.NotFound, "absent"))) };
|
||||
var ds = Source(fake);
|
||||
(await ds.ReadRawAsync("T", Now.AddMinutes(-1), Now, 10, Ct)).Samples.ShouldBeEmpty();
|
||||
(await ds.ReadRawAsync("T", Now.AddMinutes(-1), Now, 10, Ct)).Samples.ShouldBeEmpty();
|
||||
fake.TagExistsCallCount.ShouldBe(1);
|
||||
fake.ReadRawCallCount.ShouldBe(0);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task The_presence_cache_is_bounded_and_a_displaced_entry_is_checked_again()
|
||||
{
|
||||
var fake = new FakeHistorianGatewayClient { TagExistsResult = false }; var ds = Source(fake);
|
||||
for (var i = 0; i <= GatewayTagPresenceCache.Capacity; i++)
|
||||
await ds.ReadRawAsync($"T{i}", Now.AddMinutes(-1), Now, 10, Ct);
|
||||
await ds.ReadRawAsync("T0", Now.AddMinutes(-1), Now, 10, Ct);
|
||||
fake.TagExistsCallCount.ShouldBe(GatewayTagPresenceCache.Capacity + 2);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Caller_cancellation_is_not_cached_as_missing()
|
||||
{
|
||||
var fake = new FakeHistorianGatewayClient { TagExistsThrows = new OperationCanceledException() }; var ds = Source(fake);
|
||||
await Should.ThrowAsync<OperationCanceledException>(() => ds.ReadRawAsync("T", Now.AddMinutes(-1), Now, 10, Ct));
|
||||
fake.TagExistsThrows = null;
|
||||
await ds.ReadRawAsync("T", Now.AddMinutes(-1), Now, 10, Ct);
|
||||
fake.TagExistsCallCount.ShouldBe(2);
|
||||
}
|
||||
|
||||
[Theory]
|
||||
[InlineData(StatusCode.Unavailable)]
|
||||
[InlineData(StatusCode.PermissionDenied)]
|
||||
[InlineData(StatusCode.Internal)]
|
||||
public async Task A_metadata_fault_is_not_cached_as_a_missing_tag(StatusCode code)
|
||||
{
|
||||
var fake = new FakeHistorianGatewayClient { TagExistsThrows = new RpcException(new Status(code, "failure")) };
|
||||
var ds = Source(fake);
|
||||
await Should.ThrowAsync<Exception>(() => ds.ReadRawAsync("T", Now.AddMinutes(-1), Now, 10, Ct));
|
||||
fake.TagExistsThrows = null;
|
||||
await ds.ReadRawAsync("T", Now.AddMinutes(-1), Now, 10, Ct);
|
||||
fake.TagExistsCallCount.ShouldBe(2);
|
||||
fake.ReadRawCallCount.ShouldBe(1);
|
||||
}
|
||||
}
|
||||
+24
@@ -57,6 +57,30 @@ public sealed class GatewayStrictCeilingTests
|
||||
public override DateTimeOffset GetUtcNow() => new(now);
|
||||
}
|
||||
|
||||
private sealed class MovingClock : TimeProvider
|
||||
{
|
||||
public DateTime Current = Now;
|
||||
public override DateTimeOffset GetUtcNow() => new(Current);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Aggregate_ceiling_re_probes_wait_thirty_seconds_then_notice_a_raised_limit()
|
||||
{
|
||||
var fake = Store(100, ceiling: 50);
|
||||
var clock = new MovingClock();
|
||||
var ds = new GatewayHistorianDataSource(fake, NullLogger<GatewayHistorianDataSource>.Instance, timeProvider: clock);
|
||||
Task<HistoryReadResult> Read() => ds.ReadProcessedAsync("T", At(0), At(100), TimeSpan.FromSeconds(100), HistoryAggregateType.Count, Ct);
|
||||
await Should.ThrowAsync<HistoryReadLimitExceededException>(Read);
|
||||
await Should.ThrowAsync<HistoryReadLimitExceededException>(Read);
|
||||
Refusals(fake).ShouldBe(1);
|
||||
fake.MaxValuesCeiling = 200;
|
||||
await Should.ThrowAsync<HistoryReadLimitExceededException>(Read);
|
||||
Refusals(fake).ShouldBe(1);
|
||||
clock.Current = Now.AddSeconds(30);
|
||||
(await Read()).Samples.Single().Value.ShouldBe(100);
|
||||
ds.MaxRawValuesPerRead.ShouldBe(200 - RawAggregateReader.PreWindowAllowance);
|
||||
}
|
||||
|
||||
// ---- raw, uncapped ---------------------------------------------------------------------------------------------------
|
||||
|
||||
[Fact]
|
||||
|
||||
+90
@@ -0,0 +1,90 @@
|
||||
using System.Net;
|
||||
using Grpc.Core;
|
||||
using Microsoft.AspNetCore.Builder;
|
||||
using Microsoft.AspNetCore.Hosting;
|
||||
using Microsoft.AspNetCore.Server.Kestrel.Core;
|
||||
using Microsoft.Extensions.DependencyInjection;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using Microsoft.Extensions.Logging.Abstractions;
|
||||
using Shouldly;
|
||||
using Xunit;
|
||||
using ZB.MOM.WW.HistorianGateway.Contracts.Grpc;
|
||||
using ZB.MOM.WW.OtOpcUa.Runtime.Historian;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests;
|
||||
|
||||
public sealed class GatewayTagCatalogWireTests
|
||||
{
|
||||
private static CancellationToken Ct => TestContext.Current.CancellationToken;
|
||||
|
||||
[Theory]
|
||||
[InlineData("Pump1", true)]
|
||||
[InlineData("Missing", false)]
|
||||
[InlineData("Pump*", false)]
|
||||
public async Task Presence_is_proved_by_exact_names_without_using_the_broken_metadata_parser(string tag, bool expected)
|
||||
{
|
||||
var service = new TagsService();
|
||||
await using var server = await StartAsync(service);
|
||||
using var adapter = Create(server.Port);
|
||||
(await adapter.TagExistsAsync(tag, Ct)).ShouldBe(expected);
|
||||
service.Filter.ShouldBe(tag);
|
||||
service.MetadataCalls.ShouldBe(0);
|
||||
service.Completed.ShouldBe(1);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task A_catalog_stream_fault_after_a_matching_name_is_not_cached_as_presence()
|
||||
{
|
||||
var service = new TagsService { FailAfterRow = true };
|
||||
await using var server = await StartAsync(service);
|
||||
using var adapter = Create(server.Port);
|
||||
var source = new GatewayHistorianDataSource(adapter, NullLogger<GatewayHistorianDataSource>.Instance);
|
||||
var now = DateTime.UtcNow;
|
||||
await Should.ThrowAsync<Exception>(() => source.ReadRawAsync("Pump1", now.AddMinutes(-1), now, 10, Ct));
|
||||
service.FailAfterRow = false; service.Names = [];
|
||||
(await source.ReadRawAsync("Pump1", now.AddMinutes(-1), now, 10, Ct)).Samples.ShouldBeEmpty();
|
||||
service.BrowseCalls.ShouldBe(2);
|
||||
}
|
||||
|
||||
private static HistorianGatewayClientAdapter Create(int port) => HistorianGatewayClientAdapter.Create(
|
||||
new ServerHistorianOptions { Endpoint = $"http://127.0.0.1:{port}", ApiKey = "histgw_test_key", UseTls = false },
|
||||
NullLoggerFactory.Instance);
|
||||
|
||||
private sealed class TagsService : HistorianTags.HistorianTagsBase
|
||||
{
|
||||
public int MetadataCalls, BrowseCalls, Completed;
|
||||
public string? Filter;
|
||||
public bool FailAfterRow;
|
||||
public string[] Names = ["Pump1", "Pump2"];
|
||||
public override Task<GetTagMetadataReply> GetTagMetadata(GetTagMetadataRequest request, ServerCallContext context)
|
||||
{
|
||||
MetadataCalls++;
|
||||
throw new RpcException(new Status(StatusCode.Internal, "GetTagInfo response ended unexpectedly."));
|
||||
}
|
||||
public override async Task BrowseTagNames(BrowseTagNamesRequest request, IServerStreamWriter<TagName> stream, ServerCallContext context)
|
||||
{
|
||||
BrowseCalls++; Filter = request.Filter;
|
||||
foreach (var name in Names)
|
||||
{
|
||||
await stream.WriteAsync(new TagName { Name = name }, context.CancellationToken);
|
||||
if (FailAfterRow) throw new RpcException(new Status(StatusCode.Internal, "catalog failure"));
|
||||
}
|
||||
Completed++;
|
||||
}
|
||||
}
|
||||
|
||||
private sealed class Server(WebApplication app, int port) : IAsyncDisposable
|
||||
{
|
||||
public int Port => port;
|
||||
public async ValueTask DisposeAsync() { await app.StopAsync(); await app.DisposeAsync(); }
|
||||
}
|
||||
private static async Task<Server> StartAsync(TagsService service)
|
||||
{
|
||||
var builder = WebApplication.CreateSlimBuilder();
|
||||
builder.Logging.ClearProviders();
|
||||
builder.WebHost.ConfigureKestrel(k => k.Listen(IPAddress.Loopback, 0, o => o.Protocols = HttpProtocols.Http2));
|
||||
builder.Services.AddGrpc(); builder.Services.AddSingleton(service);
|
||||
var app = builder.Build(); app.MapGrpcService<TagsService>(); await app.StartAsync();
|
||||
return new Server(app, new Uri(app.Urls.First()).Port);
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user