From efaf3435ea1bd4ba6eb739a57361f282f6fe7b24 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Fri, 13 Nov 2026 01:31:52 -0500 Subject: [PATCH] fix(historian): gateway AGGREGATE_NOT_SUPPORTED_FOR_TAG -> BadAggregateNotSupported, no health failure (#900) A node declared numeric whose AVEVA tag is discrete (or string) is not refused up front by #897, so its native TimeAverage / Total still reach AVEVA, which refuses them. A HistorianGateway at or after historiangw#135 (a220ec0) answers that FailedPrecondition with a google.rpc.ErrorInfo (reason AGGREGATE_NOT_SUPPORTED_FOR_TAG, domain historian_gateway.v1) in grpc-status-details-bin. The Gateway backend's native processed read now maps exactly that to HistoryAggregateNotSupportedException (BadAggregateNotSupported for the node) without recording a health failure or treating it as transport. Any other FailedPrecondition and every Internal (an older gateway's answer for the same case) keep today's handling. Detection lives in one place, GatewayErrorReasons. The HistorianGateway.Client pin stays 1.0.0 (1.1.0 is not on the feed): 1.0.0 keeps the gateway's RpcException, trailers included, as the InnerException of the HistorianGatewayException it throws, so the reason is read with Grpc.StatusProto 2.76.0 (added via central package management; nuget.org; same version as Grpc.Core.Api; depends on Google.Api.CommonProtos 2.16.0, which accepts Google.Protobuf 3.34.1). With Client 1.1.0, ReasonOf becomes `(ex as HistorianGatewayException)?.ErrorReason` and the StatusProto reference can go. Tests: - GatewayAggregateNotSupportedWireTests: in-process gRPC gateway sending the real status + ErrorInfo shape (copied from historiangw HistorianRpcFaults.ToRpcException), read through the PUBLISHED Client 1.0.0 via the production adapter: TimeAverage / Total -> HistoryAggregateNotSupportedException, 0 health failures; controls (Internal; FailedPrecondition without ErrorInfo; other reason; other domain; the reason on Internal) are rethrown, not IOException, and counted; positive control served. - DiscreteTagTimeWeightedWireTests: over a real SDK session, a Double / Int32 node on the discrete tag -> BadAggregateNotSupported with 0 failures; the older gateway's Internal answer keeps its status and counts. - AvevaGatewayAggregateRefusalLiveGateTests: opt-in read-only live gate (OTOPCUA_AVEVA_GATE_900=true). Mutations (catch removed; status ignored; domain ignored; reason ignored; inner-exception walk removed) each fail at least one test. Release build 0 warnings; Historian.Gateway.Tests 601/607 (6 skipped), OpcUaServer IntegrationTests 323/361 (38 skipped). Live, read-only: private HistorianGateway container from a220ec0 (API-key auth disabled, live writes off, removed after) against wonder-sql-vd03: Double and Int32 nodes on SysPulse, TimeAverage and Total -> BadAggregateNotSupported, health failures unchanged (0); SysTimeSec TimeAverage Good (2 rows); SysPulse Average served. Gateway traffic was ReadAggregate / ReadRaw / GetTagMetadata only. Claude-Session: https://claude.ai/code/session_01GT67ACmpMR7b9LGRPLijMi --- Directory.Packages.props | 3 + docs/Historian.md | 11 +- .../GatewayErrorReasons.cs | 72 ++++++ .../GatewayHistorianDataSource.cs | 17 +- ...WW.OtOpcUa.Driver.Historian.Gateway.csproj | 2 + .../GatewayAggregateNotSupportedWireTests.cs | 210 ++++++++++++++++++ ...evaGatewayAggregateRefusalLiveGateTests.cs | 111 +++++++++ .../DiscreteTagTimeWeightedWireTests.cs | 75 +++++++ 8 files changed, 495 insertions(+), 6 deletions(-) create mode 100644 src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/GatewayErrorReasons.cs create mode 100644 tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayAggregateNotSupportedWireTests.cs create mode 100644 tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/AvevaGatewayAggregateRefusalLiveGateTests.cs diff --git a/Directory.Packages.props b/Directory.Packages.props index 3821cda8d..d84109317 100644 --- a/Directory.Packages.props +++ b/Directory.Packages.props @@ -32,6 +32,9 @@ + + diff --git a/docs/Historian.md b/docs/Historian.md index 960e6659a..dca6b314d 100644 --- a/docs/Historian.md +++ b/docs/Historian.md @@ -1592,9 +1592,14 @@ serve a Boolean node's numeric aggregates as 0/1 (stepped for the time-weighted true), and the Gateway's raw-evaluated Average/Min/Max/… do the same. So on a Boolean node TimeAverage / Total are served by `SqlServer` / `QuestDb` and refused by `Gateway`. An unknown node type (BaseDataType, an array) is not refused. -The residual case — a node declared numeric whose AVEVA tag is discrete — still reaches AVEVA and answers -`BadHistoryOperationUnsupported`, counted as a failure, until the gateway surfaces AVEVA's native error code as a -deterministic status. +The residual case — a node declared numeric whose AVEVA tag is discrete (or string) — still reaches AVEVA. A gateway at or +after historiangw#135 (`a220ec0`) answers it `FailedPrecondition` with a `google.rpc.ErrorInfo` (reason +`AGGREGATE_NOT_SUPPORTED_FOR_TAG`, domain `historian_gateway.v1`) in `grpc-status-details-bin`; the Gateway backend maps that +to **`BadAggregateNotSupported`** for the node and counts **no** backend failure (#900, live-verified read-only against +wonder-sql-vd03 `SysPulse`). Only that reason is matched: any other `FailedPrecondition` and every `Internal` keep the old +handling, so an older gateway (which answers this case `Internal`) still gives `BadHistoryOperationUnsupported`, counted as a +failure. The reason is read in one place, `GatewayErrorReasons` — from the trailer via `Grpc.StatusProto` while the +HistorianGateway Client pin is 1.0.0; with Client 1.1.0 it becomes `HistorianGatewayException.ErrorReason`. ⚠️ **`Server/ServerCapabilities/AggregateFunctions` (`i=2997`) is deliberately NOT trimmed** — it keeps the SDK's full 37 in every configuration, including with no historian. Part 5 defines it as every aggregate the *server* supports, and the server diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/GatewayErrorReasons.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/GatewayErrorReasons.cs new file mode 100644 index 000000000..2f268fbda --- /dev/null +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/GatewayErrorReasons.cs @@ -0,0 +1,72 @@ +using Grpc.Core; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway; + +/// +/// The HistorianGateway's machine-readable failure reasons (historiangw#135): a google.rpc.ErrorInfo the gateway +/// attaches in the standard grpc-status-details-bin trailer, domain . The ONE place OtOpcUa +/// reads them (lmxopcua#900). +/// +/// +/// The Client pin is 1.0.0, which does not surface the reason, but keeps the gateway's — +/// trailers included — as the of the HistorianGatewayException it throws, so +/// parses the trailer with Grpc.StatusProto. When the pin moves to Client 1.1.0 (not yet on the +/// feed), becomes the one line +/// (ex as ZB.MOM.WW.HistorianGateway.Client.HistorianGatewayException)?.ErrorReason (and the constants can point +/// at HistorianGatewayErrorReasons); the Grpc.StatusProto reference can then go. +/// +internal static class GatewayErrorReasons +{ + /// The ErrorInfo.domain of every reason the gateway sends (its proto package). + public const string Domain = "historian_gateway.v1"; + + /// + /// AVEVA refused the aggregate retrieval mode for this tag (TimeWeightedAverage / Integral / … on a discrete or string + /// tag; native CustomError 85 at StartQuery). Sent with . + /// + public const string AggregateNotSupportedForTag = "AGGREGATE_NOT_SUPPORTED_FOR_TAG"; + + /// + /// The gateway's ErrorInfo.reason on (or an inner exception — the client package wraps the + /// ), or when it sent none, sent one from another domain, or sent an + /// unparseable trailer. + /// + /// A gateway call's exception. + /// The reason string, or . + public static string? ReasonOf(Exception ex) + { + for (var e = ex; e is not null; e = e.InnerException) + { + if (e is not RpcException rpc) continue; + try + { + var info = rpc.GetRpcStatus()?.GetDetail(); + return info is not null && string.Equals(info.Domain, Domain, StringComparison.Ordinal) ? info.Reason : null; + } + catch (Google.Protobuf.InvalidProtocolBufferException) + { + return null; // a malformed trailer is no reason: the failure keeps its ordinary handling + } + } + return null; + } + + /// + /// True when is the gateway's refusal of an aggregate for the tag: status + /// AND reason . FailedPrecondition alone + /// is not enough — the gateway also uses it for disabled live writes and passes historian-side ones through — and an + /// older gateway answers this case Internal with no reason (unchanged handling). + /// + /// A gateway call's exception. + /// Whether it is the aggregate refusal. + public static bool IsAggregateNotSupportedForTag(Exception ex) => + StatusOf(ex) == StatusCode.FailedPrecondition + && string.Equals(ReasonOf(ex), AggregateNotSupportedForTag, StringComparison.Ordinal); + + private static StatusCode? StatusOf(Exception ex) + { + for (var e = ex; e is not null; e = e.InnerException) + if (e is RpcException rpc) return rpc.StatusCode; + return null; + } +} 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 c8da37419..8b47ba858 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 @@ -497,7 +497,7 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis var mode = AggregateModeMapper.ToRetrievalMode(aggregate); if (!_enableNativeTimeWeighted) throw new HistoryAggregateNotSupportedException(aggregate); // Planned OUTSIDE the failure-recording try: too many intervals is a client request limit, not a backend fault. - return await ReadNativeAsync(fullReference, BucketPlan.Create(startUtc, endUtc, interval), mode, cancellationToken) + return await ReadNativeAsync(fullReference, BucketPlan.Create(startUtc, endUtc, interval), aggregate, mode, cancellationToken) .ConfigureAwait(false); } @@ -505,8 +505,8 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis /// A native TimeAverage / Total read on the SQL backends' interval grid, either direction: forward native /// requests only, the clipped interval read over exactly its span. See . /// - private async Task ReadNativeAsync(string fullReference, BucketPlan plan, RetrievalMode mode, - CancellationToken cancellationToken) + private async Task ReadNativeAsync(string fullReference, BucketPlan plan, HistoryAggregateType aggregate, + RetrievalMode mode, CancellationToken cancellationToken) { if (plan.BucketCount == 0) return new HistoryReadResult([], ContinuationPoint: null); // Planned OUTSIDE the failure-recording try: a non-whole-ms interval over too many intervals is a client request limit. @@ -538,6 +538,17 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis { throw; // caller-requested, not a backend fault } + catch (Exception ex) when (GatewayErrorReasons.IsAggregateNotSupportedForTag(ex)) + { + // lmxopcua#900: AVEVA refused this retrieval mode for the tag (a node declared numeric whose AVEVA tag is discrete + // or string — the residual of #897's up-front Boolean refusal), answered by a gateway at or after historiangw#135 + // as FailedPrecondition + ErrorInfo AGGREGATE_NOT_SUPPORTED_FOR_TAG. It is the tag's configuration, not a backend + // fault: BadAggregateNotSupported for the node, no health failure, not transport. An older gateway answers the same + // case Internal with no reason, which still falls through to the generic handler below. + _logger.LogDebug("HistorianGateway refused the {Aggregate} aggregate for the tag (AGGREGATE_NOT_SUPPORTED_FOR_TAG).", + aggregate); + throw new HistoryAggregateNotSupportedException(aggregate); + } catch (Exception ex) { RecordReadFailure(ex); diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.csproj b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.csproj index c1a1d9410..e8b575bec 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.csproj +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.csproj @@ -25,6 +25,8 @@ + + diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayAggregateNotSupportedWireTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayAggregateNotSupportedWireTests.cs new file mode 100644 index 000000000..8e897c76d --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayAggregateNotSupportedWireTests.cs @@ -0,0 +1,210 @@ +using System.Net; +using Google.Protobuf; +using Google.Protobuf.WellKnownTypes; +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.Core.Abstractions; +using ZB.MOM.WW.OtOpcUa.Runtime.Historian; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests; + +/// +/// lmxopcua#900 on the wire: an in-process gRPC gateway answers ReadAggregate the way a HistorianGateway at or +/// after historiangw#135 (a220ec0) does when AVEVA refuses the retrieval mode for the tag — +/// FailedPrecondition, detail HistorianAggregateNotSupportedException, and a google.rpc.Status holding a +/// google.rpc.ErrorInfo (reason AGGREGATE_NOT_SUPPORTED_FOR_TAG, domain historian_gateway.v1, metadata +/// retrieval_mode + native_error_code) in grpc-status-details-bin (shape copied from the gateway's +/// HistorianRpcFaults.ToRpcException). Read through the PUBLISHED Client 1.0.0 (via the production adapter), so this +/// also pins that 1.0.0 keeps the trailer on the exception it throws. Controls: every other shape keeps today's handling +/// (rethrown, counted as a backend failure). +/// +public sealed class GatewayAggregateNotSupportedWireTests +{ + private static readonly DateTime S0 = new(2026, 11, 13, 12, 0, 0, DateTimeKind.Utc); + private static CancellationToken Ct => TestContext.Current.CancellationToken; + + [Theory] + [InlineData(HistoryAggregateType.TimeAverage, RetrievalMode.TimeWeightedAverage)] + [InlineData(HistoryAggregateType.Total, RetrievalMode.Integral)] + public async Task The_gateways_aggregate_refusal_is_BadAggregateNotSupported_and_no_health_failure( + HistoryAggregateType aggregate, RetrievalMode expectedMode) + { + var read = new ReadService { Fault = Refusal }; + await using var server = await StartServerAsync(read); + var (adapter, ds) = Create(server.Port); + using var _ = adapter; + + var ex = await Should.ThrowAsync(() => + ds.ReadProcessedAsync("SysPulse", S0, S0.AddMinutes(4), TimeSpan.FromMinutes(2), aggregate, Ct)); + + ex.Aggregate.ShouldBe(aggregate); + read.Modes.ShouldBe([expectedMode], "the refusal came from the gateway, not up front"); + var health = ds.GetHealthSnapshot(); + health.TotalFailures.ShouldBe(0, "a tag that cannot serve the aggregate is configuration, not a backend failure"); + health.ConsecutiveFailures.ShouldBe(0); + health.LastError.ShouldBeNull(); + } + + public static TheoryData OrdinaryFailures() => + [ + "internal", // an older gateway / a genuine historian fault: Internal, no ErrorInfo + "failed-precondition-bare", // FailedPrecondition without ErrorInfo (e.g. a historian-side one passed through) + "failed-precondition-other", // FailedPrecondition with another reason + "failed-precondition-domain", // the reason string, but not the gateway's domain + "internal-with-reason", // the reason on a status that is not FailedPrecondition + ]; + + [Theory] + [MemberData(nameof(OrdinaryFailures))] + public async Task Every_other_failure_shape_keeps_todays_handling_and_counts(string shape) + { + var read = new ReadService + { + Fault = shape switch + { + "internal" => () => new RpcException(new Status(StatusCode.Internal, "HistorianServerOperationException")), + "failed-precondition-bare" => () => new RpcException(new Status(StatusCode.FailedPrecondition, "HistorianServerRejectionException")), + "failed-precondition-other" => () => WithErrorInfo(StatusCode.FailedPrecondition, "LIVE_WRITES_DISABLED", "historian_gateway.v1"), + "failed-precondition-domain" => () => WithErrorInfo(StatusCode.FailedPrecondition, "AGGREGATE_NOT_SUPPORTED_FOR_TAG", "example.com"), + "internal-with-reason" => () => WithErrorInfo(StatusCode.Internal, "AGGREGATE_NOT_SUPPORTED_FOR_TAG", "historian_gateway.v1"), + _ => throw new ArgumentOutOfRangeException(nameof(shape)), + }, + }; + await using var server = await StartServerAsync(read); + var (adapter, ds) = Create(server.Port); + using var _ = adapter; + + var ex = await Should.ThrowAsync(() => + ds.ReadProcessedAsync("SysPulse", S0, S0.AddMinutes(4), TimeSpan.FromMinutes(2), HistoryAggregateType.TimeAverage, Ct)); + + ex.ShouldNotBeOfType(); + ex.ShouldNotBeOfType("a refusal is not a transport failure"); + ds.GetHealthSnapshot().TotalFailures.ShouldBe(1, $"{shape} is a backend failure, as before #900"); + } + + [Fact] + public async Task Positive_control_a_served_aggregate_succeeds_over_the_same_server() + { + var read = new ReadService(); + await using var server = await StartServerAsync(read); + var (adapter, ds) = Create(server.Port); + using var _ = adapter; + + var result = await ds.ReadProcessedAsync("SysTimeSec", S0, S0.AddMinutes(4), TimeSpan.FromMinutes(2), + HistoryAggregateType.TimeAverage, Ct); + + result.Samples.Count.ShouldBe(2); + ds.GetHealthSnapshot().TotalFailures.ShouldBe(0); + ds.GetHealthSnapshot().TotalSuccesses.ShouldBeGreaterThan(0); + } + + /// What a gateway at or after historiangw#135 sends (HistorianRpcFaults.ToRpcException). + private static RpcException Refusal() => new( + new Status(StatusCode.FailedPrecondition, "HistorianAggregateNotSupportedException"), + new Metadata + { + { + "grpc-status-details-bin", + new Google.Rpc.Status + { + Code = (int)StatusCode.FailedPrecondition, + Message = "HistorianAggregateNotSupportedException", + Details = + { + Any.Pack(new Google.Rpc.ErrorInfo + { + Reason = "AGGREGATE_NOT_SUPPORTED_FOR_TAG", + Domain = "historian_gateway.v1", + Metadata = { ["retrieval_mode"] = "RETRIEVAL_MODE_TIME_WEIGHTED_AVERAGE", ["native_error_code"] = "85" }, + }), + }, + }.ToByteArray() + }, + }); + + private static RpcException WithErrorInfo(StatusCode code, string reason, string domain) => new( + new Status(code, "HistorianServerRejectionException"), + new Metadata + { + { + "grpc-status-details-bin", + new Google.Rpc.Status + { + Code = (int)code, + Message = "HistorianServerRejectionException", + Details = { Any.Pack(new Google.Rpc.ErrorInfo { Reason = reason, Domain = domain }) }, + }.ToByteArray() + }, + }); + + private static (HistorianGatewayClientAdapter Adapter, GatewayHistorianDataSource Ds) Create(int port) + { + var adapter = HistorianGatewayClientAdapter.Create( + new ServerHistorianOptions { Enabled = true, Endpoint = $"http://127.0.0.1:{port}", ApiKey = "histgw_test_key", UseTls = false }, + NullLoggerFactory.Instance); + return (adapter, new GatewayHistorianDataSource(adapter, NullLogger.Instance)); + } + + private sealed class ReadService : HistorianRead.HistorianReadBase + { + private readonly List _modes = []; + + public Func? Fault { get; init; } + + public IReadOnlyList Modes { get { lock (_modes) return _modes.ToList(); } } + + public override async Task ReadAggregate(ReadAggregateRequest request, IServerStreamWriter stream, + ServerCallContext context) + { + lock (_modes) _modes.Add(request.Mode); + if (Fault is not null) throw Fault(); + var interval = request.Interval.ToTimeSpan(); + for (var t = request.StartTime.ToDateTime(); t < request.EndTime.ToDateTime(); t += interval) + await stream.WriteAsync(new HistorianAggregateSample + { + Value = 29.5, OpcQuality = 192, StartTime = Timestamp.FromDateTime(t), EndTime = Timestamp.FromDateTime(t + interval), + }); + } + } + + /// Every tag exists (GetTagMetadata answers metadata), so the read reaches ReadAggregate. + private sealed class TagsService : HistorianTags.HistorianTagsBase + { + public override Task GetTagMetadata(GetTagMetadataRequest request, ServerCallContext context) => + Task.FromResult(new GetTagMetadataReply { Metadata = new HistorianTagMetadata { Name = request.Tag } }); + } + + private sealed class Server(WebApplication app, int port) : IAsyncDisposable + { + public int Port { get; } = port; + + public async ValueTask DisposeAsync() + { + await app.StopAsync(); + await app.DisposeAsync(); + } + } + + private static async Task StartServerAsync(ReadService read) + { + 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(read); + builder.Services.AddSingleton(); + var app = builder.Build(); + app.MapGrpcService(); + app.MapGrpcService(); + await app.StartAsync(); + return new Server(app, new Uri(app.Urls.First()).Port); + } +} diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/AvevaGatewayAggregateRefusalLiveGateTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/AvevaGatewayAggregateRefusalLiveGateTests.cs new file mode 100644 index 000000000..586465d50 --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/AvevaGatewayAggregateRefusalLiveGateTests.cs @@ -0,0 +1,111 @@ +using Microsoft.Extensions.Configuration; +using Microsoft.Extensions.DependencyInjection; +using Opc.Ua; +using Opc.Ua.Client; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Commons.OpcUa; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; +using ZB.MOM.WW.OtOpcUa.Host.Configuration; +using ZB.MOM.WW.OtOpcUa.Runtime.Historian; +using static ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests.ServerEventRig; + +namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests; + +/// +/// Live gate for lmxopcua#900: a node declared numeric whose AVEVA tag is discrete (SysPulse) is not refused up +/// front (#897 refuses only Boolean nodes), so its native TimeAverage / Total reach AVEVA, which refuses them. A +/// HistorianGateway at or after historiangw#135 (a220ec0) answers FailedPrecondition + ErrorInfo +/// AGGREGATE_NOT_SUPPORTED_FOR_TAG: the booted server with the production Gateway backend must answer +/// BadAggregateNotSupported for the node and count no backend failure. +/// +/// +/// Read-only: HistoryRead only; the Gateway backend has no updater and no provisioner / recorder is built, so no +/// tag is created and nothing is written. Env: HISTGW_GATEWAY_ENDPOINT + HISTGW_GATEWAY_APIKEY (any value +/// against a gateway with API-key auth disabled) AND OTOPCUA_AVEVA_GATE_900=true (an older gateway still answers +/// Internal, which this gate would report) — skips otherwise. +/// +public sealed class AvevaGatewayAggregateRefusalLiveGateTests +{ + private const string SysPulse = "SysPulse"; + private const string SysTimeSec = "SysTimeSec"; + private static readonly DateTime S0 = new(2026, 11, 11, 14, 0, 0, DateTimeKind.Utc); + + [Fact] + public async Task A_numeric_node_on_a_discrete_AVEVA_tag_answers_BadAggregateNotSupported_without_a_health_failure() + { + var endpoint = Env("HISTGW_GATEWAY_ENDPOINT"); + var apiKey = Env("HISTGW_GATEWAY_APIKEY"); + if (endpoint is null || apiKey is null || !string.Equals(Env("OTOPCUA_AVEVA_GATE_900"), "true", StringComparison.OrdinalIgnoreCase)) + Assert.Skip("Set HISTGW_GATEWAY_ENDPOINT, HISTGW_GATEWAY_APIKEY and OTOPCUA_AVEVA_GATE_900=true (gateway at/after a220ec0)."); + + var ct = TestContext.Current.CancellationToken; + var options = new ServerHistorianOptions + { + Backend = "Gateway", Enabled = true, Endpoint = endpoint!, ApiKey = apiKey!, + UseTls = endpoint!.StartsWith("https://", StringComparison.OrdinalIgnoreCase), + }; + var provider = HistorianBackendRegistry.Resolve(options, new ConfigurationBuilder().Build()); + provider.CreateUpdater(new ServiceCollection().BuildServiceProvider()).ShouldBeNull("read-only gate"); + using var source = provider.CreateDataSource(new ServiceCollection().BuildServiceProvider()); + + await using var rig = await StartAsync("AvevaGate900", encrypted: false, ct); + rig.Nm.HistorianDataSource = source; + rig.Enforce(false, [Key()]); + var dblOnPulse = rig.EnsureRawVariable("HistDblOnPulse", "Double", writable: false, SysPulse); + var intOnPulse = rig.EnsureRawVariable("HistIntOnPulse", "Int32", writable: false, SysPulse); + var dblOnSec = rig.EnsureRawVariable("HistDblOnSec", "Double", writable: false, SysTimeSec); + + using var session = await rig.ConnectAsync(anonymous: true, ct); + var ns = (ushort)session.NamespaceUris.GetIndex(V3NodeIds.RawNamespaceUri); + + // Positive control first: the gateway is up and serves the native mode on an analog tag. + var (secStatus, secRows) = await ReadProcessedAsync(session, new NodeId(dblOnSec, ns), ObjectIds.AggregateFunction_TimeAverage, ct); + secStatus.Code.ShouldBe(StatusCodes.Good.Code, "SysTimeSec TimeAverage is served"); + secRows.ShouldBe(2); + var before = source.GetHealthSnapshot(); + before.TotalFailures.ShouldBe(0); + + foreach (var node in new[] { dblOnPulse, intOnPulse }) + foreach (var aggregate in new[] { ObjectIds.AggregateFunction_TimeAverage, ObjectIds.AggregateFunction_Total }) + { + var (status, rows) = await ReadProcessedAsync(session, new NodeId(node, ns), aggregate, ct); + status.Code.ShouldBe(StatusCodes.BadAggregateNotSupported.Code, $"{aggregate} on {node} (SysPulse)"); + rows.ShouldBe(0); + } + + // Directly on the backend: the refusal comes from AVEVA via the gateway (not up front — no type hint here). + await Should.ThrowAsync(() => source.ReadProcessedAsync( + SysPulse, S0, S0.AddMinutes(4), TimeSpan.FromMinutes(2), HistoryAggregateType.TimeAverage, ct)); + + var after = source.GetHealthSnapshot(); + after.TotalFailures.ShouldBe(before.TotalFailures, "the tag's configuration is not a backend failure"); + after.ConsecutiveFailures.ShouldBe(0); + after.LastError.ShouldBeNull(); + + // Control: the raw-evaluated Average on the same node is still served (0/1). + var (avgStatus, avgRows) = await ReadProcessedAsync(session, new NodeId(dblOnPulse, ns), ObjectIds.AggregateFunction_Average, ct); + avgStatus.Code.ShouldBe(StatusCodes.Good.Code); + avgRows.ShouldBe(2); + } + + private static async Task<(StatusCode Status, int Rows)> ReadProcessedAsync(ISession session, NodeId node, NodeId aggregate, + CancellationToken ct) + { + var details = new ReadProcessedDetails + { + StartTime = S0, EndTime = S0.AddMinutes(4), ProcessingInterval = 120_000, + AggregateType = new List { aggregate }, + AggregateConfiguration = new AggregateConfiguration { UseServerCapabilitiesDefaults = true }, + }; + var response = await session.HistoryReadAsync(null, new ExtensionObject(details), TimestampsToReturn.Both, false, + new HistoryReadValueId[] { new() { NodeId = node } }, ct); + var result = response.Results.ToList().ShouldHaveSingleItem(); + var rows = !result.HistoryData.IsNull && ExtensionObject.ToEncodeable(result.HistoryData) is HistoryData data + ? data.DataValues.Count + : 0; + return (result.StatusCode, rows); + } + + private static string? Env(string name) => Environment.GetEnvironmentVariable(name) is { Length: > 0 } v ? v.Trim() : null; +} diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/DiscreteTagTimeWeightedWireTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/DiscreteTagTimeWeightedWireTests.cs index 33b21130b..0a68347cc 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/DiscreteTagTimeWeightedWireTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/DiscreteTagTimeWeightedWireTests.cs @@ -67,6 +67,49 @@ public sealed class DiscreteTagTimeWeightedWireTests source.GetHealthSnapshot().TotalFailures.ShouldBe(0); } + /// + /// lmxopcua#900: the residual #897 case — a node declared numeric whose AVEVA tag is discrete still reaches AVEVA. A + /// gateway at or after historiangw#135 answers FailedPrecondition + ErrorInfo AGGREGATE_NOT_SUPPORTED_FOR_TAG + /// (surfaced as Client 1.0.0 surfaces it: a HistorianGatewayException wrapping the and its + /// trailers), which is BadAggregateNotSupported for the node and no health failure. Control: an older gateway's + /// Internal answer for the same read keeps today's handling (not BadAggregateNotSupported, counted). + /// + [Theory] + [InlineData("Double")] + [InlineData("Int32")] + public async Task A_numeric_node_on_a_discrete_tag_refused_by_the_gateway_is_BadAggregateNotSupported_without_a_health_failure( + string dataType) + { + var ct = TestContext.Current.CancellationToken; + await using var rig = await StartAsync("DiscreteTimeWeightedRefusal" + dataType, encrypted: false, ct); + var gateway = new AvevaShapedGateway { SendsAggregateRefusal = true }; + var source = new GatewayHistorianDataSource(gateway, NullLogger.Instance); + rig.Nm.HistorianDataSource = source; + rig.Enforce(false, [Key()]); + var node = rig.EnsureRawVariable("HistMisdeclared", dataType, writable: false, Discrete); + + using var session = await rig.ConnectAsync(anonymous: true, ct); + var ns = (ushort)session.NamespaceUris.GetIndex(V3NodeIds.RawNamespaceUri); + + foreach (var aggregate in new[] { ObjectIds.AggregateFunction_TimeAverage, ObjectIds.AggregateFunction_Total }) + { + var (status, _) = await ReadProcessedAsync(session, new NodeId(node, ns), aggregate, ct); + status.Code.ShouldBe(StatusCodes.BadAggregateNotSupported.Code, $"{aggregate} refused by the gateway for the tag"); + } + gateway.AggregateTags.Count(t => t == Discrete).ShouldBe(2, "a numeric node is not refused up front: the gateway answered"); + var health = source.GetHealthSnapshot(); + health.TotalFailures.ShouldBe(0, "the tag's configuration, not a backend failure"); + health.ConsecutiveFailures.ShouldBe(0); + health.LastError.ShouldBeNull(); + + // Control: the same read against an older gateway (Internal, no reason) keeps today's answer and counts. + gateway.SendsAggregateRefusal = false; + var (oldStatus, _) = await ReadProcessedAsync(session, new NodeId(node, ns), ObjectIds.AggregateFunction_TimeAverage, ct); + oldStatus.Code.ShouldNotBe(StatusCodes.BadAggregateNotSupported.Code); + StatusCode.IsBad(oldStatus).ShouldBeTrue(); + source.GetHealthSnapshot().TotalFailures.ShouldBe(1, "an Internal answer is still a backend failure"); + } + private static async Task<(StatusCode Status, int Rows)> ReadProcessedAsync(ISession session, NodeId node, NodeId aggregate, CancellationToken ct) { @@ -95,6 +138,12 @@ public sealed class DiscreteTagTimeWeightedWireTests { private readonly List _aggregateTags = []; + /// + /// true: answer the discrete tag's refused modes as a gateway at or after historiangw#135 does (FailedPrecondition + + /// ErrorInfo), wrapped the way Client 1.0.0's MapRpcException wraps it; false: the older Internal answer. + /// + public volatile bool SendsAggregateRefusal; + public IReadOnlyList AggregateTags { get { lock (_aggregateTags) return _aggregateTags.ToList(); } } public Task TagExistsAsync(string tag, CancellationToken ct) => Task.FromResult(true); @@ -104,6 +153,8 @@ public sealed class DiscreteTagTimeWeightedWireTests { lock (_aggregateTags) _aggregateTags.Add(tag); await Task.Yield(); + if (tag == Discrete && SendsAggregateRefusal) + throw new ZB.MOM.WW.HistorianGateway.Client.HistorianGatewayException("HistorianAggregateNotSupportedException", AggregateRefusal(mode)); if (tag == Discrete) throw new RpcException(new RpcStatus(RpcStatusCode.Internal, "HistorianServerOperationException")); for (var t = startUtc; t < endUtc; t += interval) @@ -113,6 +164,30 @@ public sealed class DiscreteTagTimeWeightedWireTests }; } + /// The gateway's refusal, shaped as historiangw HistorianRpcFaults.ToRpcException sends it (a220ec0). + private static RpcException AggregateRefusal(RetrievalMode mode) => new( + new RpcStatus(RpcStatusCode.FailedPrecondition, "HistorianAggregateNotSupportedException"), + new Grpc.Core.Metadata + { + { + "grpc-status-details-bin", + Google.Protobuf.MessageExtensions.ToByteArray(new Google.Rpc.Status + { + Code = (int)RpcStatusCode.FailedPrecondition, + Message = "HistorianAggregateNotSupportedException", + Details = + { + Any.Pack(new Google.Rpc.ErrorInfo + { + Reason = "AGGREGATE_NOT_SUPPORTED_FOR_TAG", + Domain = "historian_gateway.v1", + Metadata = { ["retrieval_mode"] = mode.ToString(), ["native_error_code"] = "85" }, + }), + }, + }) + }, + }); + public async IAsyncEnumerable ReadRawAsync(string tag, DateTime startUtc, DateTime endUtc, int maxValues, [EnumeratorCancellation] CancellationToken ct) {