feat(historian): advertise only the backend's aggregates under AggregateFunctions (#682)

IHistorianDataSource.SupportedAggregates (default: every HistoryAggregateType) is the one authority for both ServerCapabilities/AggregateFunctions (i=2997) and HistoryServerCapabilities/AggregateFunctions (i=11201) and for the up-front BadAggregateNotSupported in HistoryRead-Processed. Gateway declares AggregateModeMapper.Served (4), SqlServer/QuestDB AggregateFamilies.Served (17), Null none. Only the folder references are trimmed; the SDK's calculator factories stay so live AggregateFilter subscriptions keep working. Co-Authored-By: Claude <noreply@anthropic.com>
2026-10-12 08:52:47 -04:00
parent 8ebee2ac40
commit 6ea3ba44da
20 changed files with 654 additions and 58 deletions
@@ -129,4 +129,33 @@ public sealed class HistorianBackendRegistryTests
        o.ConnectionString.ShouldContain("Port=18812");
        o.TablePrefix.ShouldBe("rig_");
    }
+
+    /// <summary>The seventeen aggregates the SQL backends serve, spelled out (not derived) so a drift in either the
+    /// family table or a backend's declaration fails here (#682).</summary>
+    private static readonly HistoryAggregateType[] SqlServed =
+    [
+        HistoryAggregateType.Average, HistoryAggregateType.Minimum, HistoryAggregateType.Maximum, HistoryAggregateType.Count,
+        HistoryAggregateType.Range, HistoryAggregateType.Start, HistoryAggregateType.End, HistoryAggregateType.Delta,
+        HistoryAggregateType.MinimumActualTime, HistoryAggregateType.MaximumActualTime, HistoryAggregateType.TimeAverage,
+        HistoryAggregateType.TimeAverage2, HistoryAggregateType.Total, HistoryAggregateType.Total2,
+        HistoryAggregateType.Interpolative, HistoryAggregateType.PercentGood, HistoryAggregateType.PercentBad,
+    ];
+
+    [Fact]
+    public void Each_backend_data_source_declares_the_aggregates_it_serves()
+    {
+        // #682: what the server advertises under AggregateFunctions comes from these — resolved exactly as the Host does.
+        var sp = new ServiceCollection().BuildServiceProvider();
+        IHistorianDataSource Source(ServerHistorianOptions o, IConfiguration c) =>
+            HistorianBackendRegistry.Resolve(o, c).CreateDataSource(sp);
+
+        Source(Gateway(), Cfg).SupportedAggregates.ShouldBe(
+            [HistoryAggregateType.TimeAverage, HistoryAggregateType.Minimum, HistoryAggregateType.Maximum, HistoryAggregateType.Total],
+            ignoreOrder: true);
+        Source(new ServerHistorianOptions { Backend = "SqlServer", Enabled = true }, SqlCfg())
+            .SupportedAggregates.ShouldBe(SqlServed, ignoreOrder: true);
+        Source(new ServerHistorianOptions { Backend = "QuestDb", Enabled = true }, QuestDbCfg())
+            .SupportedAggregates.ShouldBe(SqlServed, ignoreOrder: true);
+        NullHistorianDataSource.Instance.SupportedAggregates.ShouldBeEmpty();
+    }
 }
@@ -0,0 +1,201 @@
+using Microsoft.Extensions.Configuration;
+using Microsoft.Extensions.DependencyInjection;
+using Opc.Ua;
+using Opc.Ua.Client;
+using Shouldly;
+using Xunit;
+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;
+
+/// <summary>
+///     Gitea #682 over a real SDK client session: <c>ServerCapabilities/AggregateFunctions</c> (i=2997) and
+///     <c>HistoryServerCapabilities/AggregateFunctions</c> (i=11201) list exactly the aggregates the wired historian
+///     backend serves — the data sources are built through <see cref="HistorianBackendRegistry"/> exactly as the Host
+///     builds them — while a live-data <c>AggregateFilter</c> subscription keeps working for every aggregate the SDK
+///     calculates, advertised or not.
+/// </summary>
+public sealed class AdvertisedAggregateFunctionsWireTests
+{
+    private static readonly string[] GatewayNames = ["Maximum", "Minimum", "TimeAverage", "Total"];
+
+    private static readonly string[] SqlNames =
+    [
+        "Average", "Count", "Delta", "End", "Interpolative", "Maximum", "MaximumActualTime", "Minimum", "MinimumActualTime",
+        "PercentBad", "PercentGood", "Range", "Start", "TimeAverage", "TimeAverage2", "Total", "Total2",
+    ];
+
+    private static IHistorianDataSource Backend(string backend)
+    {
+        var config = new ConfigurationBuilder().AddInMemoryCollection(new Dictionary<string, string?>
+        {
+            ["ServerHistorian:SqlServer:ConnectionString"] = "Server=localhost;User Id=sa;Password=x;TrustServerCertificate=True",
+            ["ServerHistorian:QuestDb:ConnectionString"] = "Host=localhost;Port=18812;Username=admin;Password=quest;Database=qdb",
+        }).Build();
+        var options = new ServerHistorianOptions
+        {
+            Backend = backend, Enabled = true, Endpoint = "http://localhost:5222", ApiKey = "histgw_id_secret", UseTls = false,
+        };
+        // Constructing a data source opens no connection; nothing here reaches a database or a gateway.
+        return HistorianBackendRegistry.Resolve(options, config).CreateDataSource(new ServiceCollection().BuildServiceProvider());
+    }
+
+    [Theory]
+    [InlineData("Gateway")]
+    [InlineData("SqlServer")]
+    [InlineData("QuestDb")]
+    public async Task Both_folders_advertise_exactly_the_backend_set(string backend)
+    {
+        var ct = TestContext.Current.CancellationToken;
+        var rig = await StartAsync($"AggAdv{backend}", encrypted: false, ct);
+        try
+        {
+            rig.Nm.HistorianDataSource = Backend(backend);
+            rig.Enforce(false, [Key()]);
+            using var session = await rig.ConnectAsync(anonymous: true, ct);
+
+            var expected = backend == "Gateway" ? GatewayNames : SqlNames;
+            (await AdvertisedAsync(session, ObjectIds.Server_ServerCapabilities_AggregateFunctions, ct)).ShouldBe(expected);
+            (await AdvertisedAsync(session, ObjectIds.HistoryServerCapabilities_AggregateFunctions, ct)).ShouldBe(expected);
+        }
+        finally { await rig.DisposeAsync(); }
+    }
+
+    [Fact]
+    public async Task With_no_historian_wired_both_folders_are_empty_from_boot()
+    {
+        // The Null default is applied by the CreateAggregateManager override, before any source is assigned.
+        var ct = TestContext.Current.CancellationToken;
+        var rig = await StartAsync("AggAdvNull", encrypted: false, ct);
+        try
+        {
+            rig.Enforce(false, [Key()]);
+            using var session = await rig.ConnectAsync(anonymous: true, ct);
+
+            (await AdvertisedAsync(session, ObjectIds.Server_ServerCapabilities_AggregateFunctions, ct)).ShouldBeEmpty();
+            (await AdvertisedAsync(session, ObjectIds.HistoryServerCapabilities_AggregateFunctions, ct)).ShouldBeEmpty();
+        }
+        finally { await rig.DisposeAsync(); }
+    }
+
+    [Fact]
+    public async Task Rewiring_the_source_resyncs_both_directions_of_the_references()
+    {
+        var ct = TestContext.Current.CancellationToken;
+        var rig = await StartAsync("AggAdvResync", encrypted: false, ct);
+        try
+        {
+            rig.Enforce(false, [Key()]);
+            using var session = await rig.ConnectAsync(anonymous: true, ct);
+
+            rig.Nm.HistorianDataSource = Backend("SqlServer");
+            (await AdvertisedAsync(session, ObjectIds.Server_ServerCapabilities_AggregateFunctions, ct)).ShouldBe(SqlNames);
+            (await ParentsAsync(session, ObjectIds.AggregateFunction_Average, ct)).ShouldBe(
+                [ObjectIds.Server_ServerCapabilities_AggregateFunctions, ObjectIds.HistoryServerCapabilities_AggregateFunctions],
+                ignoreOrder: true);
+
+            // Gateway has no arithmetic Average: dropped from both folders, and its inverse references go with it.
+            rig.Nm.HistorianDataSource = Backend("Gateway");
+            (await AdvertisedAsync(session, ObjectIds.Server_ServerCapabilities_AggregateFunctions, ct)).ShouldBe(GatewayNames);
+            (await AdvertisedAsync(session, ObjectIds.HistoryServerCapabilities_AggregateFunctions, ct)).ShouldBe(GatewayNames);
+            (await ParentsAsync(session, ObjectIds.AggregateFunction_Average, ct)).ShouldBeEmpty();
+
+            // And back: removed references are restored.
+            rig.Nm.HistorianDataSource = Backend("QuestDb");
+            (await AdvertisedAsync(session, ObjectIds.HistoryServerCapabilities_AggregateFunctions, ct)).ShouldBe(SqlNames);
+        }
+        finally { await rig.DisposeAsync(); }
+    }
+
+    [Theory]
+    [InlineData("StandardDeviationSample")] // served by no historian backend, never advertised
+    [InlineData("Average")]                 // not served by the Gateway backend wired below
+    public async Task A_live_AggregateFilter_subscription_for_an_unadvertised_aggregate_still_works(string aggregateName)
+    {
+        // The trim touches only the advertised folders, not the AggregateManager's calculator factories, which is what
+        // validates and computes a live-data AggregateFilter. Removing a factory turns this create into
+        // BadAggregateNotSupported.
+        var ct = TestContext.Current.CancellationToken;
+        var rig = await StartAsync($"AggAdvLive{aggregateName}", encrypted: false, ct);
+        try
+        {
+            rig.Nm.HistorianDataSource = Backend("Gateway");
+            rig.Enforce(false, [Key()]);
+            var id = rig.EnsureRawVariable("agg", "Double", writable: false);
+            using var session = await rig.ConnectAsync(anonymous: true, ct);
+            var aggregateId = (NodeId)typeof(ObjectIds).GetField("AggregateFunction_" + aggregateName)!.GetValue(null)!;
+
+            (await AdvertisedAsync(session, ObjectIds.Server_ServerCapabilities_AggregateFunctions, ct))
+                .ShouldNotContain(aggregateName);
+
+            var subscription = new Subscription(session.DefaultSubscription) { PublishingInterval = 100 };
+            session.AddSubscription(subscription);
+            await subscription.CreateAsync(ct);
+            var item = new MonitoredItem(subscription.DefaultItem)
+            {
+                StartNodeId = ExpandedNodeId.ToNodeId(ExpandedNodeId.Parse(RawNsu(id)), session.NamespaceUris),
+                AttributeId = Attributes.Value,
+                SamplingInterval = 100,
+                QueueSize = 10,
+                Filter = new AggregateFilter
+                {
+                    AggregateType = aggregateId,
+                    ProcessingInterval = 1000,
+                    StartTime = DateTime.UtcNow,
+                },
+            };
+            var notifications = 0;
+            item.Notification += (mi, _) => Interlocked.Add(ref notifications, mi.DequeueValues().Count);
+            subscription.AddItem(item);
+            await subscription.ApplyChangesAsync(ct);
+
+            item.Status.Error.ShouldBeNull($"AggregateFilter({aggregateName}) must be accepted on live data");
+            // And it is computed: the SDK calculator emits one value per processing interval.
+            var deadline = DateTime.UtcNow.AddSeconds(15);
+            while (Volatile.Read(ref notifications) == 0 && DateTime.UtcNow < deadline) await Task.Delay(100, ct);
+            Volatile.Read(ref notifications).ShouldBeGreaterThan(0, "the live aggregate must produce interval values");
+        }
+        finally { await rig.DisposeAsync(); }
+    }
+
+    /// <summary>Forward Organizes children of an AggregateFunctions folder, by BrowseName, sorted.</summary>
+    private static async Task<List<string>> AdvertisedAsync(ISession session, NodeId folder, CancellationToken ct)
+    {
+        var refs = await BrowseAsync(session, folder, BrowseDirection.Forward, ct);
+        return refs.Select(r => r.BrowseName.Name).OrderBy(n => n, StringComparer.Ordinal).ToList();
+    }
+
+    /// <summary>The folders an aggregate-function node is organized under (inverse Organizes).</summary>
+    private static async Task<List<NodeId>> ParentsAsync(ISession session, NodeId aggregate, CancellationToken ct)
+    {
+        var refs = await BrowseAsync(session, aggregate, BrowseDirection.Inverse, ct);
+        return refs.Select(r => ExpandedNodeId.ToNodeId(r.NodeId, session.NamespaceUris)).ToList();
+    }
+
+    private static async Task<ReferenceDescriptionCollection> BrowseAsync(
+        ISession session, NodeId node, BrowseDirection direction, CancellationToken ct)
+    {
+        var response = await session.BrowseAsync(
+            requestHeader: null,
+            view: null,
+            requestedMaxReferencesPerNode: 0,
+            nodesToBrowse:
+            [
+                new BrowseDescription
+                {
+                    NodeId = node,
+                    BrowseDirection = direction,
+                    ReferenceTypeId = ReferenceTypeIds.Organizes,
+                    IncludeSubtypes = false,
+                    NodeClassMask = 0,
+                    ResultMask = (uint)BrowseResultMask.All,
+                },
+            ],
+            ct: ct);
+        StatusCode.IsGood(response.Results[0].StatusCode).ShouldBeTrue($"browse {node}: {response.Results[0].StatusCode}");
+        return response.Results[0].References;
+    }
+}
@@ -278,6 +278,54 @@ public sealed class NodeManagerHistoryReadTests : IDisposable
        await host.DisposeAsync();
    }

+    /// <summary>
+    ///     #682: an aggregate the server maps but the wired backend does not declare in
+    ///     <see cref="IHistorianDataSource.SupportedAggregates"/> is refused <c>BadAggregateNotSupported</c> UP FRONT — the
+    ///     backend is never called (the same set is what the AggregateFunctions folders advertise). The Null default
+    ///     declares none, so a processed read with no historian is refused too.
+    /// </summary>
+    [Fact]
+    public async Task Processed_aggregate_outside_the_backend_set_is_refused_without_a_backend_call()
+    {
+        var (host, server) = await BootAsync();
+        var nm = server.NodeManager!;
+        var fake = new RecordingHistorianDataSource
+        {
+            ProcessedResult = Empty(),
+            Supported = new HashSet<HistoryAggregateType> { HistoryAggregateType.TimeAverage },
+        };
+        nm.HistorianDataSource = fake;
+
+        nm.EnsureVariable("eq-1/gw", parentFolderNodeId: null, displayName: "Gw", dataType: "Float",
+            writable: false, historianTagname: "WW.Gw", realm: AddressSpaceRealm.Uns);
+        var nodeId = nm.TryGetVariable("eq-1/gw")!.NodeId;
+
+        ReadProcessedDetails Details(NodeId aggregate) => new()
+        {
+            StartTime = DateTime.UtcNow.AddHours(-1),
+            EndTime = DateTime.UtcNow,
+            ProcessingInterval = 10_000.0,
+            AggregateType = new NodeIdCollection { aggregate },
+        };
+
+        // Average is mapped by the server but not in this backend's set ⇒ refused, no call.
+        InvokeHistoryRead(server, nm, Details(ObjectIds.AggregateFunction_Average), nodeId).Item2[0].StatusCode.Code
+            .ShouldBe(StatusCodes.BadAggregateNotSupported);
+        fake.LastCall.ShouldBeNull();
+
+        // Positive control: a member of the set reaches the backend.
+        InvokeHistoryRead(server, nm, Details(ObjectIds.AggregateFunction_TimeAverage), nodeId);
+        fake.LastCall.ShouldBe("Processed");
+        fake.LastAggregate.ShouldBe(HistoryAggregateType.TimeAverage);
+
+        // No historian: the Null default serves no aggregate.
+        nm.HistorianDataSource = NullHistorianDataSource.Instance;
+        InvokeHistoryRead(server, nm, Details(ObjectIds.AggregateFunction_TimeAverage), nodeId).Item2[0].StatusCode.Code
+            .ShouldBe(StatusCodes.BadAggregateNotSupported);
+
+        await host.DisposeAsync();
+    }
+
    /// <summary>
    ///     #677: a client <c>AggregateConfiguration</c> that differs from the server defaults is refused per node with
    ///     <c>BadAggregateConfigurationRejected</c> (it used to be silently ignored); the source is never invoked.
@@ -626,6 +674,8 @@ public sealed class NodeManagerHistoryReadTests : IDisposable
        public bool ThrowOnRead { get; init; }
        public bool AggregateNotSupported { get; init; }
        public bool LimitExceeded { get; init; }
+        public IReadOnlySet<HistoryAggregateType> Supported { get; init; } = HistoryAggregateTypes.All;
+        public IReadOnlySet<HistoryAggregateType> SupportedAggregates => Supported;
        public HistorianRead RawResult { get; set; } = new(Array.Empty<DataValueSnapshot>(), null);
        public HistorianRead ProcessedResult { get; set; } = new(Array.Empty<DataValueSnapshot>(), null);
        public HistorianRead AtTimeResult { get; set; } = new(Array.Empty<DataValueSnapshot>(), null);