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) {