From 97668d26d3f7c8156bdeefafcc095af01d48d538 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Tue, 13 Oct 2026 03:51:04 -0400 Subject: [PATCH 1/2] fix(historian): respect gateway ceilings and history value contracts --- docs/Historian.md | 20 ++- .../2026-10-13-historian-695-698-results.md | 28 ++++ .../GatewayAtTimeReader.cs | 22 ++- .../GatewayHistorianDataSource.cs | 25 +++- .../QuestDbHistorianDataSource.cs | 10 ++ .../SqlServerHistorianDataSource.cs | 8 ++ .../OtOpcUaNodeManager.cs | 59 +++++++- .../ReverseRawReader.cs | 24 ++-- .../GatewayAtTimeTests.cs | 29 ++++ .../GatewayCeilingPagingTests.cs | 48 +++++++ .../QuestDbHistorianProviderTests.cs | 27 ++++ .../SqlServerHistorianProviderTests.cs | 27 ++++ .../NodeManagerHistoryReadEventsTests.cs | 14 +- .../NodeManagerHistoryReadModifiedTests.cs | 32 ++++- .../NodeManagerHistoryReadTests.cs | 130 ++++++++++++++++++ 15 files changed, 474 insertions(+), 29 deletions(-) create mode 100644 docs/plans/2026-10-13-historian-695-698-results.md diff --git a/docs/Historian.md b/docs/Historian.md index 6c774fa9..0c4963d9 100644 --- a/docs/Historian.md +++ b/docs/Historian.md @@ -573,7 +573,8 @@ node without `isHistorized` has no HistoryRead access bit and answers | Historized node, historian configured, time range empty | `GoodNoData` | | Historized node, historian NOT configured (`Enabled=false` / Null source) | `GoodNoData` (empty) for Raw / AtTime; Processed ⇒ `BadAggregateNotSupported` (#682) | | Non-historized node | `BadHistoryOperationUnsupported` | -| Backend timeout or exception | `BadHistoryOperationUnsupported` per node; other nodes in the same batch are unaffected | +| Backend transport outage (#697) | `BadNoCommunication` per node across Raw, Modified, Processed, AtTime and Events; cancelled/deadline-limited server reads retain `BadTimeout` | +| Other backend exception / unsupported operation | `BadHistoryOperationUnsupported` per node; other nodes in the same batch are unaffected | | `Backend=SqlServer`, `AutoCreateSchema=false`, schema missing/behind | every read/write fails with a clear error naming the missing schema version (health `LastError`); reads answer `BadHistoryOperationUnsupported` per node; no host crash | A historized node with no historian configured never returns an error status — it returns @@ -823,6 +824,10 @@ data. it could only ever be a Part 11 *interpolated* bound. The start bound on the Gateway is the real prior sample, found by the #686 walk-back. +**Warm crowding ceiling (#695).** A raw crowding re-read spends the remaining room below a learned gateway ceiling before +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. + ### 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, @@ -880,9 +885,14 @@ data. here — the SQL backends store DateTime as its own kind and return a `DateTime`. Under an integer hint a NaN, ±∞ or out-of-Int64-range stored value is no integer: it is treated as unusable (as `BadNoData`), so a neighbour like that is never interpolated or saturated into a Good value. -- **What differs between backends (by design, not changed by #693).** Only the **Gateway** converts an AtTime result to the node's - integer type (`Int32` ⇒ `int`): the SQL backends store and return integer nodes as `Int64` (and Boolean as Boolean). Gateway - **Raw** reads still return a Boolean node's samples as the stored Double 0/1 (the hint is passed only on AtTime). +- **Client value types (#698, all backends).** Before returning Raw, Modified or AtTime history, the node manager converts non-null + scalar values to the node's declared DataType (including Boolean, integer widths and Float). Arrays retain their stored values. + Conversion failures produce a null value with `BadTypeMismatch` for that sample, preserving timestamps and modification metadata. + DateTime requires an actual DateTime value: the gateway's numeric DateTime encoding is unknown and cannot be inferred. + Processed results retain each aggregate's result type, so Count and averages can remain Double on a Boolean or integer node. +- **Future timestamps (#696).** A bounded raw probe checks for stored samples between now and the requested future time. An empty + future interval starts the prior-sample search near now; a full probe seeds the existing walk-back with a stored future sample. + The result still carries the requested timestamp and the evaluator's Uncertain status when no later sample exists. - ⚠️ **Shared-evaluator fix (all backends that use it):** `AtTimeEvaluator` returned every interpolated `Int64` boxed as a **Double** (a C# `cond ? (long)x : double` conditional widens the long); its "keeping the Int64 type" tests passed because Shouldly's `ShouldBe` compares numbers across types. SqlServer / QuestDB interpolated `Int64` AtTime values are now `Int64`, @@ -1764,7 +1774,7 @@ this family and **are** capped (they used to be pushed down). Raise the cap, or ### Health The read client reports `GetHealthSnapshot()` (query/failure counters, consecutive failures, `LastError` — e.g. a missing schema under -`AutoCreateSchema=false`). A SQL error on a read surfaces as `BadHistoryOperationUnsupported` for **that node only**; a cancelled read +`AutoCreateSchema=false`). A SQL connection outage on a read surfaces as `BadNoCommunication`; other SQL errors retain `BadHistoryOperationUnsupported` for **that node only**; a cancelled read (deadline, client abort) is not counted as a backend fault. ### Retention (`RetentionDays`) diff --git a/docs/plans/2026-10-13-historian-695-698-results.md b/docs/plans/2026-10-13-historian-695-698-results.md new file mode 100644 index 00000000..701ad326 --- /dev/null +++ b/docs/plans/2026-10-13-historian-695-698-results.md @@ -0,0 +1,28 @@ +# Historian follow-up results — 2026-10-13 + +Scope: approved bounded fixes for #695–#698, plus live measurement of #653. Base: `66a2b0c3`. + +- #695: spend the remaining learned-ceiling room on crowding reads; reduce crowded reverse-read windows using the observed pre-window count. Regression cases cover ceilings 1,709 and 1,369, and reverse paging through a crowded millisecond. +- #696 item 1: probe the future interval before starting the prior-sample search near now. Preserve stored future samples, requested timestamps and Uncertain status. Dense future probes still use the existing bounded walk. Items 2 and 3 remain open (stopped-tag strict-ceiling cost and aggregate ceiling re-probes). +- #697: actual Gateway, SQL Server and QuestDB transport failures become `BadNoCommunication` on every HistoryRead arm. Cancellation remains `BadTimeout`; unsupported operations retain `BadHistoryOperationUnsupported`. Tests include real refused database connections, repeated reads and nested gRPC failures. +- #698: normalize Raw, Modified and AtTime scalar values to the node DataType. Conversion failure gives per-sample null/`BadTypeMismatch`; arrays, audit metadata and timestamps survive. Processed results retain aggregate-specific types. +- #694 item 3 remains blocked on historiangw#27. No upstream code was changed. + +## Live #653 measurement + +Five sequential calls per operation and tag, using client 0.3.0 against the running local gateway (`localhost:5221`) with its existing remote AVEVA connection. Commissioned a two-hour `historian:read` key, then revoked it after measurement. No historian writes or provisioning performed. + +| Operation | SysTimeSec, elapsed ms | Nonexistent tag, elapsed ms | +|---|---|---| +| Metadata | 299, 85, 81, 82, 48 (success) | 89, 53, 41, 77, 76 (gateway exception) | +| Raw | 143, 99, 145, 133, 159 (success) | 30260, 30634, 30675, 30754, 30628 (gateway exception) | +| Aggregate | 271, 275, 309, 249, 304 (success) | 30564, 30605, 30583, 30575, 30806 (gateway exception) | +| AtTime | 140, 88, 89, 127, 92 (success) | 61074, 61330, 61601, 61249, 61271 (gateway exception) | + +Nonexistent tag: `OtOpcUa.Issue653.Nonexistent.20261013`. The delay is still present. Recommend a metadata pre-check with a short negative cache, caching only explicit missing-tag outcomes and never authentication, communication or arbitrary errors. This is follow-up implementation; #653 remains open. Direct gateway AtTime above differs from OtOpcUa's raw-based AtTime evaluator. + +## Consumer and validation notes + +ScadaBridge source and tests contain no HistoryRead consumer; no consumer change required. Independent review found a future-stored-sample edge case, which was fixed and covered before final verification. No deployment or live end-to-end OPC UA verification of these new fixes is claimed; the live run measures #653 through the gateway client. + +Full solution: 14,581 passed, 609 skipped, 0 failed across 62 test suites. Command: `dotnet test ZB.MOM.WW.OtOpcUa.slnx --verbosity quiet`, exit 0. Skips include environment-gated integration suites. diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/GatewayAtTimeReader.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/GatewayAtTimeReader.cs index f6d9f107..1827f2ff 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/GatewayAtTimeReader.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/GatewayAtTimeReader.cs @@ -227,7 +227,27 @@ internal static class GatewayAtTimeReader { if (!beforeKnown) { - before = await WalkBackAsync(ctx, qs, walkKnown, walkWidth).ConfigureAwait(false); + // #696: prove the future stretch empty before jumping to now. Historian clocks or imports + // can place real rows there; a full bounded probe seeds the original backward search. + var searchEnd = qs; + if (qs > now && walkKnown is null) + { + var future = await ReadTrustedAsync(ctx, MsFloor(now), + Min(MsFloor(qs) + OneMs, horizon), ReadRows).ConfigureAwait(false); + StoredSample? latest = null; + foreach (var row in future.Rows) + if (row.TimeUtc <= qs) latest = row; + if (future.Complete) + { + before = latest; + beforeKnown = latest is not null; + searchEnd = now; + } + else if (latest is { } found) + walkKnown = found.TimeUtc; + } + if (!beforeKnown) + before = await WalkBackAsync(ctx, searchEnd, walkKnown, walkWidth).ConfigureAwait(false); beforeKnown = true; } prev = before; diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/GatewayHistorianDataSource.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/GatewayHistorianDataSource.cs index f4cc345d..cc3ed941 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/GatewayHistorianDataSource.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/GatewayHistorianDataSource.cs @@ -152,6 +152,7 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis catch (Exception ex) { RecordReadFailure(ex); + if (IsTransportFailure(ex)) throw new System.IO.IOException("HistorianGateway read transport is unavailable.", ex); throw; } } @@ -239,7 +240,15 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis for (var attempt = 0; attempt < MaxCrowdedReReads && preWindow <= _maxPreWindowRows; attempt++) { var budget = (int)Math.Min(int.MaxValue, (long)cap + room + RawAggregateReader.PreWindowAllowance); - if (_ceiling.Learned is var known and > 0 && budget > known) break; // over a strict ceiling: refused anyway + // #695: 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 - cap - RawAggregateReader.PreWindowAllowance; + if (available <= 0) break; + room = Math.Min(room, available); + budget = cap + 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 pass.Result.Samples; preWindow = Math.Max(preWindow, pass.Result.PreWindowRows); @@ -365,6 +374,7 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis catch (Exception ex) { RecordReadFailure(ex); + if (IsTransportFailure(ex)) throw new System.IO.IOException("HistorianGateway read transport is unavailable.", ex); throw; } } @@ -408,6 +418,7 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis catch (Exception ex) { RecordReadFailure(ex); + if (IsTransportFailure(ex)) throw new System.IO.IOException("HistorianGateway read transport is unavailable.", ex); throw; } } @@ -466,6 +477,7 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis catch (Exception ex) { RecordReadFailure(ex); + if (IsTransportFailure(ex)) throw new System.IO.IOException("HistorianGateway read transport is unavailable.", ex); throw; } } @@ -486,6 +498,7 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis catch (Exception ex) { RecordReadFailure(ex); + if (IsTransportFailure(ex)) throw new System.IO.IOException("HistorianGateway read transport is unavailable.", ex); throw; } } @@ -581,6 +594,7 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis catch (Exception ex) { RecordReadFailure(ex); + if (IsTransportFailure(ex)) throw new System.IO.IOException("HistorianGateway read transport is unavailable.", ex); throw; } } @@ -655,6 +669,15 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis return result; } + /// Classifies actual gRPC/package transport failures without treating refusals as outages (#697). + 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), + }; + /// /// 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. diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianDataSource.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianDataSource.cs index 9d7e9a6e..ed77f5f9 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianDataSource.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianDataSource.cs @@ -137,6 +137,7 @@ public sealed class QuestDbHistorianDataSource : IHistorianDataSource catch (Exception ex) { RecordFailure(ex); + if (IsTransportFailure(ex)) throw new System.IO.IOException("Historian read transport is unavailable.", ex); throw; } } @@ -232,6 +233,7 @@ public sealed class QuestDbHistorianDataSource : IHistorianDataSource catch (Exception ex) { RecordFailure(ex); + if (IsTransportFailure(ex)) throw new System.IO.IOException("Historian read transport is unavailable.", ex); throw; } } @@ -316,6 +318,7 @@ public sealed class QuestDbHistorianDataSource : IHistorianDataSource catch (Exception ex) { RecordFailure(ex); + if (IsTransportFailure(ex)) throw new System.IO.IOException("Historian read transport is unavailable.", ex); throw; } } @@ -443,6 +446,7 @@ public sealed class QuestDbHistorianDataSource : IHistorianDataSource catch (Exception ex) { RecordFailure(ex); + if (IsTransportFailure(ex)) throw new System.IO.IOException("Historian read transport is unavailable.", ex); throw; } } @@ -538,6 +542,7 @@ public sealed class QuestDbHistorianDataSource : IHistorianDataSource catch (Exception ex) { RecordFailure(ex); + if (IsTransportFailure(ex)) throw new System.IO.IOException("Historian read transport is unavailable.", ex); throw; } } @@ -701,6 +706,11 @@ public sealed class QuestDbHistorianDataSource : IHistorianDataSource private void RecordSuccess() => _health.RecordSuccess(); + /// Classifies provider transport/timeouts, keeping SQL/query errors distinct (#697). + private static bool IsTransportFailure(Exception ex) => ex is NpgsqlException { IsTransient: true } and not PostgresException + || ex is PostgresException pg && (pg.SqlState.StartsWith("08", StringComparison.Ordinal) + || pg.SqlState is "57P01" or "57P02" or "57P03"); + private void RecordFailure(Exception ex) { // A cancelled read (deadline / client abort) is not a backend fault (the counters ignore it too). diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerHistorianDataSource.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerHistorianDataSource.cs index 6fe53f93..c5cd9444 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerHistorianDataSource.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerHistorianDataSource.cs @@ -132,6 +132,7 @@ public sealed class SqlServerHistorianDataSource : IHistorianDataSource catch (Exception ex) { RecordFailure(ex); + if (IsTransportFailure(ex)) throw new System.IO.IOException("Historian read transport is unavailable.", ex); throw; } } @@ -229,6 +230,7 @@ public sealed class SqlServerHistorianDataSource : IHistorianDataSource catch (Exception ex) when (ex is not NotSupportedException) { RecordFailure(ex); + if (IsTransportFailure(ex)) throw new System.IO.IOException("Historian read transport is unavailable.", ex); throw; } } @@ -306,6 +308,7 @@ public sealed class SqlServerHistorianDataSource : IHistorianDataSource catch (Exception ex) { RecordFailure(ex); + if (IsTransportFailure(ex)) throw new System.IO.IOException("Historian read transport is unavailable.", ex); throw; } } @@ -401,6 +404,7 @@ public sealed class SqlServerHistorianDataSource : IHistorianDataSource catch (Exception ex) { RecordFailure(ex); + if (IsTransportFailure(ex)) throw new System.IO.IOException("Historian read transport is unavailable.", ex); throw; } } @@ -512,6 +516,7 @@ public sealed class SqlServerHistorianDataSource : IHistorianDataSource catch (Exception ex) { RecordFailure(ex); + if (IsTransportFailure(ex)) throw new System.IO.IOException("Historian read transport is unavailable.", ex); throw; } } @@ -635,6 +640,9 @@ public sealed class SqlServerHistorianDataSource : IHistorianDataSource private void RecordSuccess() => _health.RecordSuccess(); + /// Classifies provider transport/timeouts, keeping SQL/query errors distinct (#697). + private static bool IsTransportFailure(Exception ex) => ex is SqlException sql && sql.Number is -2 or 0 or 53 or 64 or 233 or 258 or 10053 or 10054 or 10060 or 10061 or 10065 or 11001 or 40197 or 40613; + private void RecordFailure(Exception ex) { // A cancelled read (deadline / client abort) is not a backend fault (the counters ignore it too). diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs index 3d284e44..6b0d40b6 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs @@ -5129,6 +5129,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 RunBounded(work, HistoryReadBatchParallelism); }); + NormalizeHistoryValues(nodesToProcess, results); } /// @@ -5225,6 +5226,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 RunBounded(work, HistoryReadBatchParallelism); }); + NormalizeHistoryValues(nodesToProcess, results); } /// @@ -5516,8 +5518,8 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 // One node's backend failure must not throw out of the batch — surface Bad for THIS node // only. Logged via the injected ILogger. _logger.LogError(ex, "OtOpcUaNodeManager: HistoryReadEvents failed for node {NodeId}", handle.NodeId); - errors[handle.Index] = StatusCodes.BadHistoryOperationUnsupported; - results[handle.Index] = new SdkHistoryReadResult { StatusCode = StatusCodes.BadHistoryOperationUnsupported }; + errors[handle.Index] = HistoryReadFailureStatus(ex); + results[handle.Index] = new SdkHistoryReadResult { StatusCode = HistoryReadFailureStatus(ex) }; } } @@ -6233,7 +6235,8 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 // Bad status for THIS node only. Logged through the injected ILogger // rather than swallowed silently. _logger.LogError(ex, "OtOpcUaNodeManager: HistoryRead failed for node {NodeId}", handle.NodeId); - errors[handle.Index] = StatusCodes.BadHistoryOperationUnsupported; + errors[handle.Index] = HistoryReadFailureStatus(ex); + results[handle.Index] = new SdkHistoryReadResult { StatusCode = HistoryReadFailureStatus(ex) }; } } @@ -6722,7 +6725,8 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 { // One node's backend failure must not throw out of the batch — Bad for THIS node only. _logger.LogError(ex, "OtOpcUaNodeManager: HistoryReadRaw (paged) failed for node {NodeId}", handle.NodeId); - errors[handle.Index] = StatusCodes.BadHistoryOperationUnsupported; + errors[handle.Index] = HistoryReadFailureStatus(ex); + results[handle.Index] = new SdkHistoryReadResult { StatusCode = HistoryReadFailureStatus(ex) }; } } @@ -6829,8 +6833,8 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 catch (Exception ex) { _logger.LogError(ex, "OtOpcUaNodeManager: HistoryRead (modified) failed for node {NodeId}", handle.NodeId); - errors[handle.Index] = StatusCodes.BadHistoryOperationUnsupported; - results[handle.Index] = new SdkHistoryReadResult { StatusCode = StatusCodes.BadHistoryOperationUnsupported }; + errors[handle.Index] = HistoryReadFailureStatus(ex); + results[handle.Index] = new SdkHistoryReadResult { StatusCode = HistoryReadFailureStatus(ex) }; } } @@ -6879,6 +6883,49 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 } } + /// Transport failures are retryable; unsupported operations retain their distinct status (#697). + private static uint HistoryReadFailureStatus(Exception error) + { + for (Exception? ex = error; ex is not null; ex = ex.InnerException) + if (ex is System.IO.IOException or System.Net.Sockets.SocketException or System.Net.Http.HttpRequestException or TimeoutException) + return StatusCodes.BadNoCommunication; + return StatusCodes.BadHistoryOperationUnsupported; + } + + /// Normalize stored and AtTime history to each scalar node's declared DataType (#698). + private void NormalizeHistoryValues(List handles, IList results) + { + foreach (var handle in handles) + { + if (AtTimeValueType(handle.NodeId) is not { } type || results[handle.Index]?.HistoryData is not { } body) continue; + var values = ExtensionObject.ToEncodeable(body) switch + { + HistoryData data => data.DataValues, + _ => null, + }; + if (values is null) continue; + foreach (var value in values) + { + if (value.Value is null) continue; + try + { + // A gateway number cannot reconstruct a DateTime or a String which was never stored. + if ((type == TypeCode.DateTime && value.Value is not DateTime) + || (type == TypeCode.String && value.Value is not string)) throw new InvalidCastException(); + var converted = Convert.ChangeType(value.Value, type, System.Globalization.CultureInfo.InvariantCulture); + if (converted is float f && float.IsInfinity(f) && value.Value is double d && double.IsFinite(d)) + throw new OverflowException(); + value.WrappedValue = new Variant(converted); + } + catch (Exception ex) when (ex is InvalidCastException or OverflowException or FormatException) + { + value.WrappedValue = Variant.Null; + value.StatusCode = StatusCodes.BadTypeMismatch; + } + } + } + } + /// Project a plain sample list into an SDK (the paged Raw path /// works on a trimmed rather than a whole ). /// The samples to project (already trimmed of boundary duplicates). diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/ReverseRawReader.cs b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/ReverseRawReader.cs index dda79544..93572869 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/ReverseRawReader.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/ReverseRawReader.cs @@ -100,16 +100,22 @@ internal static class ReverseRawReader { var asked = limit > 0 && cap > limit ? limit : cap; // cap 0 (the whole window) is never cut IReadOnlyList samples; - try + for (var attempt = 0; ; attempt++) { - samples = (await source.ReadRawAsync(tagname, start, end, (uint)asked, token).ConfigureAwait(false)).Samples; - } - catch (HistoryReadLimitExceededException ex) when (asked > 0 && ex.PreWindowRows > 0 && asked - ex.PreWindowRows >= 1) - { - // #694: rows just before the segment crowded the backend's budget — read it smaller; a full answer - // then counts as truncated (the walk narrows), never as complete. - asked -= ex.PreWindowRows; - samples = (await source.ReadRawAsync(tagname, start, end, (uint)asked, token).ConfigureAwait(false)).Samples; + try + { + samples = (await source.ReadRawAsync(tagname, start, end, (uint)asked, token).ConfigureAwait(false)).Samples; + break; + } + catch (HistoryReadLimitExceededException ex) when (asked > 0 && ex.PreWindowRows > 0 && attempt < 2) + { + // #695: the discovery request may be smaller than the pre-window cluster. Size against + // the ceiling, not that request; a second pass can discover more crowding rows. + var ceilingRoom = source.MaxRawValuesPerRead is var learned and > 0 ? learned : asked; + var smaller = Math.Min(asked - 1, ceilingRoom - ex.PreWindowRows - 1); + if (smaller < 1) throw; + asked = smaller; + } } return new ForwardSegmentRead(samples, asked > 0 && samples.Count >= asked); }, diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayAtTimeTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayAtTimeTests.cs index 11e47fa2..08644285 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayAtTimeTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayAtTimeTests.cs @@ -128,6 +128,35 @@ public sealed class GatewayAtTimeTests return r.Samples; } + [Fact] + public async Task A_future_time_on_a_live_tag_finds_the_prior_sample_without_walking_the_empty_future() + { + var store = new AvevaStore(Enumerable.Range(0, 7201).Select(i => (Now.AddSeconds(i - 7200), (double)i, false))); + var fake = store.Client(); + var result = (await AtTime(fake, Now.AddHours(1))).Single(); + result.Value.ShouldBe(7200d); + result.StatusCode.ShouldBe(UncertainInterpolated); + fake.ReadRawCallCount.ShouldBeLessThanOrEqualTo(4); + } + + [Fact] + public async Task A_future_time_still_uses_a_stored_future_prior_sample() + { + var fake = new AvevaStore([(Now.AddSeconds(-1), 1d, false), (Now.AddMinutes(30), 2d, false)]).Client(); + var result = (await AtTime(fake, Now.AddHours(1))).Single(); + result.Value.ShouldBe(2d); + result.StatusCode.ShouldBe(UncertainInterpolated); + } + + [Fact] + public async Task A_full_future_probe_walks_to_the_latest_stored_future_sample() + { + var fake = new AvevaStore(Enumerable.Range(0, 7201).Select(i => (Now.AddSeconds(i), (double)i, false))).Client(); + var result = (await AtTime(fake, Now.AddHours(3))).Single(); + result.Value.ShouldBe(7200d); + result.StatusCode.ShouldBe(UncertainInterpolated); + } + // ---- the four live findings ------------------------------------------------------------------------------------ [Fact] // live: 19:00:03 (a stored NULL) ⇒ 107 Good diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayCeilingPagingTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayCeilingPagingTests.cs index 8730b5c9..0ca4db5d 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayCeilingPagingTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayCeilingPagingTests.cs @@ -420,6 +420,54 @@ public sealed class GatewayCeilingPagingTests : IDisposable finally { await host.DisposeAsync(); } } + [Theory] + [InlineData(1709, 1366)] // #695 reverse-walk trace + [InlineData(1369, 1292)] // #695 bounds trace + public async Task A_warm_read_uses_the_last_room_below_the_ceiling_for_pre_window_ties(int ceiling, int cluster) + { + var start = At(1000); + var stored = Enumerable.Range(0, cluster).Select(i => (start.AddMilliseconds(-0.5), (double)i)) + .Append((start, 9000d)).Append((start.AddSeconds(1), 9001d)).ToList(); + var fake = Gateway(stored, ceiling); + using var ds = new GatewayHistorianDataSource(fake, NullLogger.Instance); + await ds.ReadRawAsync("T", At(0), At(1), 5000, Ct); // learn without crowding + fake.RawCalls.Clear(); + var result = await ds.ReadRawAsync("T", start, start.AddSeconds(2), 1, Ct); + result.Samples.Select(s => s.Value).ShouldBe([9000d]); + fake.RawCalls.ShouldAllBe(c => c.MaxValues <= ceiling); + } + + [Fact] + public async Task A_warm_reverse_segment_can_shrink_again_when_the_pre_window_cluster_grows() + { + var start = At(1000); + var stored = Enumerable.Range(0, 1366).Select(i => (start.AddMilliseconds(-0.5), (double)i)) + .Concat(Enumerable.Range(0, 500).Select(i => (start.AddMilliseconds(i), 9000d + i))).ToList(); + var fake = Gateway(stored, 1709); + using var ds = new GatewayHistorianDataSource(fake, NullLogger.Instance); + await ds.ReadRawAsync("T", At(0), At(1), 5000, Ct); + var page = await ReverseRawReader.ReadAsync(ds, "T", start.AddMilliseconds(499), start.AddTicks(-1), + 249, 65536, At(2000), new ReverseWalkHint(499 * TimeSpan.TicksPerMillisecond), Ct); + page.Samples.Select(s => (double)s.Value!).ShouldBe(Enumerable.Range(9251, 249).Reverse().Select(i => (double)i)); + } + + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task Gateway_transport_faults_surface_as_BadNoCommunication(bool deadline) + { + var fake = new FakeHistorianGatewayClient { ThrowOnRead = deadline + ? new ZB.MOM.WW.HistorianGateway.Client.HistorianGatewayException("deadline", new Grpc.Core.RpcException(new Grpc.Core.Status(Grpc.Core.StatusCode.DeadlineExceeded, "deadline"))) + : new ZB.MOM.WW.HistorianGateway.Client.HistorianGatewayUnavailableException("unavailable") }; + var (host, nm, nodeId) = await BootAsync(fake); + try + { + var r = ReadRaw(nm, nodeId, At(0), At(10), 10, null); + r.Error.StatusCode.Code.ShouldBe(StatusCodes.BadNoCommunication); + } + finally { await host.DisposeAsync(); } + } + // ---- helpers --------------------------------------------------------------------------------------------------------- private async Task<(OpcUaApplicationHost Host, OtOpcUaNodeManager Nm, NodeId NodeId)> BootAsync( diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.Tests/QuestDbHistorianProviderTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.Tests/QuestDbHistorianProviderTests.cs index 8f3b4d5d..c67a9b03 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.Tests/QuestDbHistorianProviderTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.Tests/QuestDbHistorianProviderTests.cs @@ -11,6 +11,33 @@ public sealed class QuestDbHistorianProviderTests ConnectionString = "Host=localhost;Port=18812;Username=admin;Password=quest;Database=qdb;ServerCompatibilityMode=NoTypeLoading", }); + [Theory] + [InlineData("Raw")] + [InlineData("AtTime")] + [InlineData("Processed")] + [InlineData("Events")] + [InlineData("Modified")] + public async Task A_connection_failure_is_exposed_as_a_transport_failure(string kind) + { + using var provider = new QuestDbHistorianProvider(new QuestDbHistorianOptions { + ConnectionString = "Host=127.0.0.1;Port=1;Username=admin;Password=test;Database=qdb;Timeout=1;Pooling=False;ServerCompatibilityMode=NoTypeLoading", AutoCreateSchema = false + }); + using var source = provider.CreateDataSource(new NullSp()); + var t = new DateTime(2026, 1, 1, 0, 0, 0, DateTimeKind.Utc); + Func read = kind switch + { + "Raw" => async () => await source.ReadRawAsync("T", t, t.AddHours(1), 10, TestContext.Current.CancellationToken), + "AtTime" => async () => await source.ReadAtTimeAsync("T", [t], TestContext.Current.CancellationToken), + "Processed" => async () => await source.ReadProcessedAsync("T", t, t.AddHours(1), TimeSpan.FromSeconds(10), HistoryAggregateType.Average, TestContext.Current.CancellationToken), + "Events" => async () => await source.ReadEventsAsync("T", t, t.AddHours(1), 10, TestContext.Current.CancellationToken), + _ => async () => await source.ReadModifiedAsync("T", t, t.AddHours(1), 10, null, TestContext.Current.CancellationToken) + }; + var error = await Should.ThrowAsync(read); + error.InnerException.ShouldNotBeNull(); + // A cached bootstrap outage must remain retryable on subsequent reads as well. + (await Should.ThrowAsync(read)).InnerException.ShouldNotBeNull(); + } + [Fact] public void Name_and_all_six_kinds() { diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.Tests/SqlServerHistorianProviderTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.Tests/SqlServerHistorianProviderTests.cs index 9c8cb24d..f3ab5176 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.Tests/SqlServerHistorianProviderTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.Tests/SqlServerHistorianProviderTests.cs @@ -11,6 +11,33 @@ public sealed class SqlServerHistorianProviderTests ConnectionString = "Server=localhost;User Id=sa;Password=x;TrustServerCertificate=True", }); + [Theory] + [InlineData("Raw")] + [InlineData("AtTime")] + [InlineData("Processed")] + [InlineData("Events")] + [InlineData("Modified")] + public async Task A_connection_failure_is_exposed_as_a_transport_failure(string kind) + { + var provider = new SqlServerHistorianProvider(new SqlServerHistorianOptions { + ConnectionString = "Server=127.0.0.1,1;User Id=sa;Password=test;TrustServerCertificate=True;Connect Timeout=1;ConnectRetryCount=0;Pooling=False", AutoCreateSchema = false + }); + using var source = provider.CreateDataSource(new NullSp()); + var t = new DateTime(2026, 1, 1, 0, 0, 0, DateTimeKind.Utc); + Func read = kind switch + { + "Raw" => async () => await source.ReadRawAsync("T", t, t.AddHours(1), 10, TestContext.Current.CancellationToken), + "AtTime" => async () => await source.ReadAtTimeAsync("T", [t], TestContext.Current.CancellationToken), + "Processed" => async () => await source.ReadProcessedAsync("T", t, t.AddHours(1), TimeSpan.FromSeconds(10), HistoryAggregateType.Average, TestContext.Current.CancellationToken), + "Events" => async () => await source.ReadEventsAsync("T", t, t.AddHours(1), 10, TestContext.Current.CancellationToken), + _ => async () => await source.ReadModifiedAsync("T", t, t.AddHours(1), 10, null, TestContext.Current.CancellationToken) + }; + var error = await Should.ThrowAsync(read); + error.InnerException.ShouldNotBeNull(); + // A cached bootstrap outage must remain retryable on subsequent reads as well. + (await Should.ThrowAsync(read)).InnerException.ShouldNotBeNull(); + } + [Fact] public void Name_and_all_six_kinds() { diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadEventsTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadEventsTests.cs index e49759a7..36679959 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadEventsTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadEventsTests.cs @@ -295,12 +295,14 @@ public sealed class NodeManagerHistoryReadEventsTests : IDisposable /// HistoryRead call. The fake source MUST be invoked (proving we reached the bridge) and threw; /// the production catch now sets BOTH errors and results explicitly, so both are asserted here /// rather than relying on the SDK base pre-seeding results[i]. - [Fact] - public async Task Events_backend_throw_yields_bad_status_and_does_not_escape() + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task Events_backend_throw_yields_bad_status_and_does_not_escape(bool communication) { var (host, server) = await BootAsync(); var nm = server.NodeManager!; - var fake = new RecordingHistorianDataSource { ThrowOnRead = true }; + var fake = new RecordingHistorianDataSource { ThrowOnRead = !communication, ReadException = communication ? new System.IO.IOException("lost connection") : null }; nm.HistorianDataSource = fake; const string equipmentId = "eq-boom"; @@ -321,10 +323,10 @@ public sealed class NodeManagerHistoryReadEventsTests : IDisposable // Authoritative per-node signal: errors[0] must be Bad. ServiceResult.IsBad(errors[0]).ShouldBeTrue(); - errors[0].StatusCode.Code.ShouldBe(StatusCodes.BadHistoryOperationUnsupported); + errors[0].StatusCode.Code.ShouldBe(communication ? StatusCodes.BadNoCommunication : StatusCodes.BadHistoryOperationUnsupported); // The production catch now sets results[0] explicitly — assert it directly (not relying on base seeding). results[0].ShouldNotBeNull(); - results[0].StatusCode.Code.ShouldBe(StatusCodes.BadHistoryOperationUnsupported); + results[0].StatusCode.Code.ShouldBe(communication ? StatusCodes.BadNoCommunication : StatusCodes.BadHistoryOperationUnsupported); // The source WAS entered (proving the override reached the bridge and the throw was swallowed). fake.EventsEntered.ShouldBeTrue(); // ThrowOnRead fires before EventsCalled/LastSourceName are set — that's the throw path. @@ -380,6 +382,7 @@ public sealed class NodeManagerHistoryReadEventsTests : IDisposable private sealed class RecordingHistorianDataSource : IHistorianDataSource { public bool ThrowOnRead { get; init; } + public Exception? ReadException { get; init; } public HistoricalEventsResult EventsResult { get; set; } = new(Array.Empty(), null); @@ -411,6 +414,7 @@ public sealed class NodeManagerHistoryReadEventsTests : IDisposable CancellationToken cancellationToken) { EventsEntered = true; + if (ReadException is not null) throw ReadException; if (ThrowOnRead) throw new InvalidOperationException("backend boom for events"); EventsCalled = true; LastSourceName = sourceName; diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadModifiedTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadModifiedTests.cs index 48b10136..1e62ebc8 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadModifiedTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadModifiedTests.cs @@ -25,6 +25,34 @@ public sealed class NodeManagerHistoryReadModifiedTests : IDisposable private static readonly DateTime Epoch = new(2026, 1, 1, 0, 0, 0, DateTimeKind.Utc); private readonly string _pkiRoot = Path.Combine(Path.GetTempPath(), $"otopcua-modified-{Guid.NewGuid():N}"); + [Fact] + public async Task Modified_values_use_the_scalar_node_type_and_preserve_the_audit_information() + { + var (host, nm, node, _) = await SetupAsync([Mod(1, 1, 2d, HistoryModificationType.Replace, "alice", Epoch)], "Float"); + try + { + var r = Read(nm, node, Epoch, Epoch.AddHours(1), 10, null); + r.Data!.DataValues.Single().Value.ShouldBeOfType().ShouldBe(2f); + r.Data.ModificationInfos.Single().UserName.ShouldBe("alice"); + r.Data.ModificationInfos.Single().UpdateType.ShouldBe(HistoryUpdateType.Replace); + } + finally { await host.DisposeAsync(); } + } + + [Fact] + public async Task A_modified_read_of_an_unreachable_backend_answers_BadNoCommunication() + { + var (host, nm, node, src) = await SetupAsync([]); + try + { + src.Throw = new System.IO.IOException("connection lost"); + var r = Read(nm, node, Epoch, Epoch.AddHours(1), 10, null); + r.Error.StatusCode.Code.ShouldBe(StatusCodes.BadNoCommunication); + r.Result.StatusCode.Code.ShouldBe(StatusCodes.BadNoCommunication); + } + finally { await host.DisposeAsync(); } + } + [Fact] public async Task A_modified_read_dispatches_to_the_trail_and_projects_modification_infos() { @@ -289,7 +317,7 @@ public sealed class NodeManagerHistoryReadModifiedTests : IDisposable } private async Task<(OpcUaApplicationHost Host, OtOpcUaNodeManager Nm, NodeId Node, SqlLikeTrail Src)> SetupAsync( - List trail) + List trail, string dataType = "Double") { var started = await OpcUaHostStart.StartAsync( port => new OpcUaApplicationHost( @@ -307,7 +335,7 @@ public sealed class NodeManagerHistoryReadModifiedTests : IDisposable nm.HistoryContinuationStore = new InMemoryHistoryContinuationStore(); var src = new SqlLikeTrail(trail); nm.HistorianDataSource = src; - nm.EnsureVariable("eq-1/mod", null, "Mod", "Double", false, historianTagname: Tag, realm: AddressSpaceRealm.Uns); + nm.EnsureVariable("eq-1/mod", null, "Mod", dataType, false, historianTagname: Tag, realm: AddressSpaceRealm.Uns); return (started.Host, nm, nm.TryGetVariable("eq-1/mod")!.NodeId, src); } diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadTests.cs index 0d314f6d..aa83493f 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadTests.cs @@ -786,6 +786,132 @@ public sealed class NodeManagerHistoryReadTests : IDisposable return (results, errors); } + [Theory] + [InlineData("Raw")] + [InlineData("AtTime")] + [InlineData("Processed")] + public async Task An_unreachable_backend_answers_a_communication_status(string kind) + { + var (host, server) = await BootAsync(); + try + { + var nm = server.NodeManager!; + nm.HistorianDataSource = new RecordingHistorianDataSource { ReadException = new System.IO.IOException("connection lost") }; + nm.EnsureVariable("eq/comm", null, "Comm", "Double", false, historianTagname: "T", realm: AddressSpaceRealm.Uns); + var t = DateTime.UtcNow.AddMinutes(-1); + HistoryReadDetails details = kind switch + { + "Raw" => new ReadRawModifiedDetails { IsReadModified = false, ReturnBounds = false, StartTime = t, EndTime = t.AddSeconds(10) }, + "AtTime" => new ReadAtTimeDetails { ReqTimes = new DateTimeCollection { t } }, + _ => new ReadProcessedDetails { StartTime = t, EndTime = t.AddSeconds(10), ProcessingInterval = 1000, + AggregateType = new NodeIdCollection { ObjectIds.AggregateFunction_Average } } + }; + var (results, errors) = InvokeHistoryRead(server, nm, details, nm.TryGetVariable("eq/comm")!.NodeId); + errors[0].StatusCode.Code.ShouldBe(StatusCodes.BadNoCommunication); + results[0].StatusCode.Code.ShouldBe(StatusCodes.BadNoCommunication); + } + finally { await host.DisposeAsync(); } + } + + [Theory] + [InlineData("Int32", "Raw")] + [InlineData("Boolean", "Raw")] + [InlineData("Float", "Raw")] + [InlineData("Int32", "AtTime")] + [InlineData("Boolean", "AtTime")] + [InlineData("Float", "AtTime")] + public async Task History_values_use_the_nodes_scalar_type(string dataType, string kind) + { + var (host, server) = await BootAsync(); + try + { + var nm = server.NodeManager!; + var t = DateTime.UtcNow.AddMinutes(-1); + var value = new HistorianRead([new DataValueSnapshot(1d, StatusCodes.Good, t, t)], null); + nm.HistorianDataSource = new RecordingHistorianDataSource { RawResult = value, AtTimeResult = value }; + nm.EnsureVariable("eq/typed", null, "Typed", dataType, false, historianTagname: "T", realm: AddressSpaceRealm.Uns); + HistoryReadDetails details = kind == "Raw" + ? new ReadRawModifiedDetails { IsReadModified = false, ReturnBounds = false, StartTime = t, EndTime = t.AddSeconds(10) } + : new ReadAtTimeDetails { ReqTimes = new DateTimeCollection { t } }; + var (results, errors) = InvokeHistoryRead(server, nm, details, nm.TryGetVariable("eq/typed")!.NodeId); + errors[0].StatusCode.Code.ShouldBe(StatusCodes.Good); + var actual = ((HistoryData)ExtensionObject.ToEncodeable(results[0].HistoryData)).DataValues.Single(); + if (dataType == "Int32") actual.Value.ShouldBeOfType().ShouldBe(1); + if (dataType == "Boolean") actual.Value.ShouldBeOfType().ShouldBeTrue(); + if (dataType == "Float") actual.Value.ShouldBeOfType().ShouldBe(1f); + actual.SourceTimestamp.ShouldBe(t); + actual.StatusCode.Code.ShouldBe(StatusCodes.Good); + } + finally { await host.DisposeAsync(); } + } + + [Theory] + [InlineData("Int16", 40000d)] + [InlineData("DateTime", 1d)] + public async Task An_unrepresentable_historical_value_is_null_with_BadTypeMismatch(string type, double input) + { + var (host, server) = await BootAsync(); + try + { + var nm = server.NodeManager!; + var t = DateTime.UtcNow.AddMinutes(-1); + nm.EnsureVariable("eq/mismatch", null, "Mismatch", type, false, historianTagname: "T", realm: AddressSpaceRealm.Uns); + nm.HistorianDataSource = new RecordingHistorianDataSource { RawResult = new HistorianRead([new(input, StatusCodes.Good, t, t)], null) }; + var (results, _) = InvokeHistoryRead(server, nm, new ReadRawModifiedDetails { IsReadModified = false, ReturnBounds = false, + StartTime = t, EndTime = t.AddSeconds(10) }, nm.TryGetVariable("eq/mismatch")!.NodeId); + var value = ((HistoryData)ExtensionObject.ToEncodeable(results[0].HistoryData)).DataValues.Single(); + value.Value.ShouldBeNull(); + value.StatusCode.Code.ShouldBe(StatusCodes.BadTypeMismatch); + value.SourceTimestamp.ShouldBe(t); + value.ServerTimestamp.ShouldBe(t); + } + finally { await host.DisposeAsync(); } + } + + [Fact] + public async Task Array_and_null_history_values_preserve_quality_and_timestamps() + { + var (host, server) = await BootAsync(); + try + { + var nm = server.NodeManager!; + var t = DateTime.UtcNow.AddMinutes(-1); + var serverTime = t.AddSeconds(1); + nm.EnsureVariable("eq/array", null, "Array", "Int32", false, historianTagname: "T", realm: AddressSpaceRealm.Raw, isArray: true, arrayLength: 2); + nm.HistorianDataSource = new RecordingHistorianDataSource { RawResult = new HistorianRead([ + new(new int[] { 1, 2 }, StatusCodes.Uncertain, t, serverTime), + new(null, StatusCodes.BadNoCommunication, t.AddSeconds(1), serverTime)], null) }; + var (results, _) = InvokeHistoryRead(server, nm, new ReadRawModifiedDetails { IsReadModified = false, ReturnBounds = false, + StartTime = t, EndTime = t.AddSeconds(10) }, nm.TryGetVariable("eq/array", AddressSpaceRealm.Raw)!.NodeId); + var values = ((HistoryData)ExtensionObject.ToEncodeable(results[0].HistoryData)).DataValues; + values[0].Value.ShouldBeOfType().ShouldBe([1, 2]); + values[0].StatusCode.Code.ShouldBe(StatusCodes.Uncertain); + values[0].SourceTimestamp.ShouldBe(t); + values[0].ServerTimestamp.ShouldBe(serverTime); + values[1].Value.ShouldBeNull(); + values[1].StatusCode.Code.ShouldBe(StatusCodes.BadNoCommunication); + } + finally { await host.DisposeAsync(); } + } + + [Fact] + public async Task Processed_Count_keeps_its_aggregate_type_on_a_Boolean_node() + { + var (host, server) = await BootAsync(); + try + { + var nm = server.NodeManager!; + var t = DateTime.UtcNow.AddMinutes(-1); + nm.EnsureVariable("eq/count", null, "Count", "Boolean", false, historianTagname: "T", realm: AddressSpaceRealm.Uns); + nm.HistorianDataSource = new RecordingHistorianDataSource { ProcessedResult = new HistorianRead([new(3d, StatusCodes.Good, t, t)], null) }; + var (results, _) = InvokeHistoryRead(server, nm, new ReadProcessedDetails { StartTime = t, EndTime = t.AddSeconds(10), + ProcessingInterval = 1000, AggregateType = new NodeIdCollection { ObjectIds.AggregateFunction_Count } }, nm.TryGetVariable("eq/count")!.NodeId); + var value = ((HistoryData)ExtensionObject.ToEncodeable(results[0].HistoryData)).DataValues.Single(); + value.Value.ShouldBeOfType().ShouldBe(3d); + } + finally { await host.DisposeAsync(); } + } + private static HistorianRead Empty() => new(Array.Empty(), null); /// A recording fake historian source — captures the last call's kind + arguments and returns @@ -793,6 +919,7 @@ public sealed class NodeManagerHistoryReadTests : IDisposable private sealed class RecordingHistorianDataSource : IHistorianDataSource { public bool ThrowOnRead { get; init; } + public Exception? ReadException { get; init; } public bool AggregateNotSupported { get; init; } public bool LimitExceeded { get; init; } public IReadOnlySet? Supported { get; init; } = HistoryAggregateTypes.All; @@ -814,6 +941,7 @@ public sealed class NodeManagerHistoryReadTests : IDisposable string fullReference, DateTime startUtc, DateTime endUtc, uint maxValuesPerNode, CancellationToken cancellationToken) { + if (ReadException is not null) throw ReadException; if (ThrowOnRead) throw new InvalidOperationException("backend boom"); LastCall = "Raw"; LastTagname = fullReference; @@ -827,6 +955,7 @@ public sealed class NodeManagerHistoryReadTests : IDisposable string fullReference, DateTime startUtc, DateTime endUtc, TimeSpan interval, HistoryAggregateType aggregate, CancellationToken cancellationToken) { + if (ReadException is not null) throw ReadException; if (ThrowOnRead) throw new InvalidOperationException("backend boom"); if (AggregateNotSupported) throw new HistoryAggregateNotSupportedException(aggregate); if (LimitExceeded) throw new HistoryReadLimitExceededException(4, "too many raw rows"); @@ -842,6 +971,7 @@ public sealed class NodeManagerHistoryReadTests : IDisposable public Task ReadAtTimeAsync( string fullReference, IReadOnlyList timestampsUtc, CancellationToken cancellationToken) { + if (ReadException is not null) throw ReadException; if (ThrowOnRead) throw new InvalidOperationException("backend boom"); LastCall = "AtTime"; LastTagname = fullReference; -- 2.52.0 From 58977640245d1ef7c95bc0d37c6007e586fce725 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Tue, 13 Oct 2026 04:31:52 -0400 Subject: [PATCH 2/2] fix(historian): fail fast for missing tags and reduce gateway read cost --- docs/Historian.md | 18 ++- .../2026-10-13-historian-653-696-results.md | 36 +++++ .../GatewayHistorianDataSource.cs | 35 ++++- .../GatewayReadLimits.cs | 27 +++- .../GatewayTagPresenceCache.cs | 36 +++++ .../HistorianGatewayClientAdapter.cs | 11 ++ .../IHistorianGatewayClient.cs | 6 + .../RawAggregateReader.cs | 2 +- .../ReverseRawReader.cs | 2 +- .../ReverseWalk.cs | 18 ++- .../FakeHistorianGatewayClient.cs | 11 ++ .../GatewayCeilingPagingTests.cs | 15 +++ .../GatewayMissingTagTests.cs | 126 ++++++++++++++++++ .../GatewayStrictCeilingTests.cs | 24 ++++ .../GatewayTagCatalogWireTests.cs | 90 +++++++++++++ 15 files changed, 439 insertions(+), 18 deletions(-) create mode 100644 docs/plans/2026-10-13-historian-653-696-results.md create mode 100644 src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/GatewayTagPresenceCache.cs create mode 100644 tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayMissingTagTests.cs create mode 100644 tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayTagCatalogWireTests.cs diff --git a/docs/Historian.md b/docs/Historian.md index 0c4963d9..e6801b31 100644 --- a/docs/Historian.md +++ b/docs/Historian.md @@ -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, diff --git a/docs/plans/2026-10-13-historian-653-696-results.md b/docs/plans/2026-10-13-historian-653-696-results.md new file mode 100644 index 00000000..df052676 --- /dev/null +++ b/docs/plans/2026-10-13-historian-653-696-results.md @@ -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. diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/GatewayHistorianDataSource.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/GatewayHistorianDataSource.cs index cc3ed941..854bfda5 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/GatewayHistorianDataSource.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/GatewayHistorianDataSource.cs @@ -68,7 +68,8 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis internal int MaxPreWindowRows => _maxPreWindowRows; /// A strict gateway's row ceiling, learned from its first refusal (#691.3). - 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); } /// @@ -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(), t => AggregateEvaluator.NoData(t, _time.GetUtcNow().UtcDateTime)), ContinuationPoint: null); + } var byBucket = new Dictionary(); 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(), 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; } + /// No-data values for a missing tag, stamped with the client's requested times. + private static HistoryReadResult MissingAtTime(IReadOnlyList times, DateTime now) => + new(times.Select(t => AtTimeEvaluator.Evaluate(t, null, null, now, now)).ToArray(), ContinuationPoint: null); + /// Classifies actual gRPC/package transport failures without treating refusals as outages (#697). private static bool IsTransportFailure(Exception ex) => ex switch { diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/GatewayReadLimits.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/GatewayReadLimits.cs index a4467136..56951345 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/GatewayReadLimits.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/GatewayReadLimits.cs @@ -76,13 +76,15 @@ internal static class GatewayReadLimits /// the read is re-issued at N; refused naming C again ⇒ the C-bounded result stands (no repeat). /// /// -/// Why no TTL. A raise only matters to a read that needs more than 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 BadTooManyOperations whatever is done. +/// Re-probe timing (#696). 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. /// /// -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 /// The learned ceiling (0 = none). @@ -93,8 +95,9 @@ internal sealed class GatewayRowCeiling /// The max_values the read wants. /// One drained read with the given budget. /// True when a pass with the given budget delivered it all without proving completeness. + /// Aggregate-only opt-in: reuse a refusal for 30 seconds before re-probing an incomplete read. /// The result and the budget it was read with. - public async Task<(T Result, int Budget)> ReadAsync(int budget, Func> pass, Func needsMore) + public async Task<(T Result, int Budget)> ReadAsync(int budget, Func> pass, Func 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(int budget, Func> 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); } } diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/GatewayTagPresenceCache.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/GatewayTagPresenceCache.cs new file mode 100644 index 00000000..19760334 --- /dev/null +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/GatewayTagPresenceCache.cs @@ -0,0 +1,36 @@ +using Grpc.Core; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway; + +/// Bounded metadata cache (#653). Missing tags expire quickly; failures never become missing tags. +internal sealed class GatewayTagPresenceCache(IHistorianGatewayClient client, TimeProvider time) +{ + internal const int Capacity = 4096; + private readonly object _sync = new(); + private readonly Dictionary _entries = new(StringComparer.Ordinal); + + public async Task 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; + } +} diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/HistorianGatewayClientAdapter.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/HistorianGatewayClientAdapter.cs index c7a323d1..b4ceefd3 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/HistorianGatewayClientAdapter.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/HistorianGatewayClientAdapter.cs @@ -81,6 +81,17 @@ public sealed class HistorianGatewayClientAdapter : IHistorianGatewayClient, IDi return new HistorianGatewayClientAdapter(HistorianGatewayClient.Create(clientOptions)); } + /// + public async Task 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 + } + /// public IAsyncEnumerable ReadRawAsync( string tag, DateTime startUtc, DateTime endUtc, int maxValues, CancellationToken ct) => diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/IHistorianGatewayClient.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/IHistorianGatewayClient.cs index eef687c0..f356d52d 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/IHistorianGatewayClient.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/IHistorianGatewayClient.cs @@ -11,6 +11,12 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway; /// public interface IHistorianGatewayClient : IAsyncDisposable { + /// Checks the tag catalog; false means explicitly absent, while failures propagate. + /// Historian tag name. + /// Cancellation token. + /// Whether the historian contains the tag. + Task TagExistsAsync(string tag, CancellationToken ct); + /// Streams raw historian samples for a tag over a time window. /// The historian tag name to read. /// Inclusive start of the requested time window (UTC). diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/RawAggregateReader.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/RawAggregateReader.cs index 6f144c77..7f64e36f 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/RawAggregateReader.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/RawAggregateReader.cs @@ -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); diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/ReverseRawReader.cs b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/ReverseRawReader.cs index 93572869..5491bb4f 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/ReverseRawReader.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/ReverseRawReader.cs @@ -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 }; } diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/ReverseWalk.cs b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/ReverseWalk.cs index 014c760d..79816766 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/ReverseWalk.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/ReverseWalk.cs @@ -123,13 +123,14 @@ internal static class ReverseWalk /// What the previous page of the same read carried (default = none). /// Builds the exception for a timestamp holding more than M rows. /// Cancellation. + /// Use ceiling-sized discovery and retain truncated remainder rows (#696, strict raw backends only). /// At most rows, newest first, and the hint for the next page. public static async Task<(List Rows, ReverseWalkHint Next)> WalkAsync( Func>> read, Func ticksOf, IComparer? tieOrder, DateTime hi, DateTime lo, uint want, int tieBudget, long resolutionTicks, DateTime nowUtc, - ReverseWalkHint hint, Func tooManyAtOneTimestamp, CancellationToken ct) + ReverseWalkHint hint, Func 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 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); diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/FakeHistorianGatewayClient.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/FakeHistorianGatewayClient.cs index 5ee232b0..80decb3f 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/FakeHistorianGatewayClient.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/FakeHistorianGatewayClient.cs @@ -12,6 +12,17 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests; /// public sealed class FakeHistorianGatewayClient : IHistorianGatewayClient { + public bool TagExistsResult = true; + public int TagExistsCallCount; + public Exception? TagExistsThrows; + public Task TagExistsAsync(string tag, CancellationToken ct) + { + ct.ThrowIfCancellationRequested(); + TagExistsCallCount++; + if (TagExistsThrows is { } error) throw error; + return Task.FromResult(TagExistsResult); + } + // ---- ReadRaw ------------------------------------------------------------------------------- public IReadOnlyList RawSamples = Array.Empty(); public string? LastReadRawTag; diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayCeilingPagingTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayCeilingPagingTests.cs index 0ca4db5d..17c5d312 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayCeilingPagingTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayCeilingPagingTests.cs @@ -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.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] diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayMissingTagTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayMissingTagTests.cs new file mode 100644 index 00000000..992ae6fd --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayMissingTagTests.cs @@ -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.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(() => 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(() => 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); + } +} diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayStrictCeilingTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayStrictCeilingTests.cs index 2dbd50ae..5b6f0e80 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayStrictCeilingTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayStrictCeilingTests.cs @@ -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.Instance, timeProvider: clock); + Task Read() => ds.ReadProcessedAsync("T", At(0), At(100), TimeSpan.FromSeconds(100), HistoryAggregateType.Count, Ct); + await Should.ThrowAsync(Read); + await Should.ThrowAsync(Read); + Refusals(fake).ShouldBe(1); + fake.MaxValuesCeiling = 200; + await Should.ThrowAsync(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] diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayTagCatalogWireTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayTagCatalogWireTests.cs new file mode 100644 index 00000000..23fb00f2 --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayTagCatalogWireTests.cs @@ -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.Instance); + var now = DateTime.UtcNow; + await Should.ThrowAsync(() => 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 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 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 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(); await app.StartAsync(); + return new Server(app, new Uri(app.Urls.First()).Port); + } +} -- 2.52.0