From 6ea3ba44da9c9c6ab1c69f366d27303c4e167857 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Mon, 12 Oct 2026 08:52:47 -0400 Subject: [PATCH 1/3] 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 --- CLAUDE.md | 3 +- docs/Historian.md | 50 ++++- .../Historian/HistoryReadContracts.cs | 12 ++ .../Historian/IHistorianDataSource.cs | 17 ++ .../Historian/NullHistorianDataSource.cs | 9 + .../GatewayHistorianDataSource.cs | 7 + .../Mapping/AggregateModeMapper.cs | 31 ++- .../QuestDbHistorianDataSource.cs | 4 + .../AggregateFamily.cs | 9 + .../SqlServerHistorianDataSource.cs | 4 + .../AdvertisedAggregateFunctions.cs | 91 ++++++++ .../AggregateFunctionMap.cs | 55 +++++ .../OtOpcUaNodeManager.cs | 86 ++++---- .../OtOpcUaSdkServer.cs | 15 ++ ...torianDataSourceReverseEventsGuardTests.cs | 15 ++ .../Mapping/AggregateModeMapperTests.cs | 15 ++ .../ProcessedReadTests.cs | 9 + .../HistorianBackendRegistryTests.cs | 29 +++ .../AdvertisedAggregateFunctionsWireTests.cs | 201 ++++++++++++++++++ .../NodeManagerHistoryReadTests.cs | 50 +++++ 20 files changed, 654 insertions(+), 58 deletions(-) create mode 100644 src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/AdvertisedAggregateFunctions.cs create mode 100644 src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/AggregateFunctionMap.cs create mode 100644 tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/AdvertisedAggregateFunctionsWireTests.cs diff --git a/CLAUDE.md b/CLAUDE.md index 0d6f9d5b..08a6f7a0 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -1353,6 +1353,7 @@ historian tagname, which is the tag's driver `FullName`. - **SQL Server** (`Driver.Historian.SqlServer`): `hist.Sample` PK `(TagId, SourceTimeUtc)` — writes are insert-if-absent, so outbox redelivery is idempotent. Stores every value kind with its **real status**; per-sample aggregates (10, simple definitions) are pushed down to T-SQL with bucket math in **100 ns ticks**; the Part 13 time-weighted seven (TimeAverage/TimeAverage2/Total/Total2/Interpolative, #671 — ⚠️ `Total` is now the time-weighted integral, no longer `SUM` — plus the **duration-based** PercentGood/PercentBad, #680) stream window + 4 bound rows into the shared `TimeWeightedAggregator` (both SQL backends, identical results; reverse `Interpolative` anchored at the later time; window capped by `ServerHistorian:MaxTimeWeightedRawRows` ⇒ `BadTooManyOperations`, percentages included). ⚠️ **#680 status rules (client-visible):** TimeAverage/Total are Uncertain if any part of the interval uses an Uncertain value, is bridged across a Bad sample, is extrapolated past the last sample (⇒ a live tag's latest interval is Uncertain) or is not covered; TimeAverage2/Total2 keep the SDK time-based rule (Uncertain time counts as Good) but never extrapolate past the last sample — that interval is Partial, later ones `BadNoData` (a report-by-exception tag with nothing after its last sample reads `BadNoData` until its next sample arrives). `docs/Historian.md` §"Time-weighted aggregates". ⚠️ **#680 per-sample (client-visible):** `Count` counts **Good** samples only (value-based status; an all-Bad interval is generic Bad, value null) and an empty interval **inside the tag's data range** is `0`/Good (outside it still `BadNoData`); `Start`/`End` return the **raw** first/last sample of any quality — its own timestamp and status, value null when Bad; `Delta` is Uncertain only when the interval's first or last sample is unusable; and every per-sample aggregate but `Average` sets **end-of-data Partial** on the interval holding the last sample (forward incl. its earlier edge, reverse strictly inside), never on a Bad result. The data-range lookup is a seek on SQL Server but a partition scan on QuestDB (unindexed `tag` SYMBOL). `docs/Historian.md` §"Aggregates". Docker-dev recipe, schema and aggregate semantics: `docs/Historian.md` §"SQL Server backend". Integration tests need `HISTORIAN_SQL_TEST_CONNECTION`. **Retention (#667):** opt-in `SqlServer:RetentionDays` (+ `PurgeInterval` 1h, `PurgeBatchSize` 4000) -> `HistorianRetentionService` purges old `hist.Sample` (per tag) / `hist.Event` in `DELETE TOP(n)` batches, Primary-gated on `IHistorizationRoleView`, one purger per DB via `sp_getapplock`; counter `otopcua.historian.retention.purged{table}`. Optional `IHistorianBackendProvider.CreateRetention` (default null). `docs/Historian.md` §Retention. - **QuestDB** (`Driver.Historian.QuestDb`, `ServerHistorian:QuestDb:*`, Npgsql over PG wire; docker-dev overlay `docker-compose.questdb.yml`): same kinds/status/aggregates/HistoryUpdate as SqlServer, soft delete by tombstone (INSERT only — never `UPDATE`/`DELETE`; every read filters `NOT deleted`). ⚠️ **Timestamps are µs** (`QuestDbConnection.Round`/`BindTimestamp`; `IHistorianDataSource.TimestampResolutionTicks` = 10 steps HistoryRead paging), **pre-1970 cannot be stored** (writer drops, HistoryUpdate `BadOutOfRange`), every write waits for WAL visibility, TTL is enforced on write commits only, set Npgsql `Maximum Pool Size` < QuestDB's 64. `docs/Historian.md` §"QuestDB backend". - ⚠️ **Reverse processed reads are Part 13/SDK-anchored (#677, client-visible):** every aggregate on SqlServer/QuestDB now anchors a reverse read at its later time — window `(end, start]`, partial interval earliest, each stamped with its later edge (was: earlier-anchored, stamped earlier edge); on forward reads Partial also marks the interval the data starts inside (not Average — the SDK's rule, forward-only); a non-default client `AggregateConfiguration` ⇒ `BadAggregateConfigurationRejected`; Gateway provisioning sends `IntegralDivisor = 1`. SDK-parity tests: `SdkAggregateParityTests`. `docs/Historian.md` §Aggregates. +- **Advertised aggregates (#682):** `IHistorianDataSource.SupportedAggregates` (Gateway 4 via `AggregateModeMapper.Served`, SqlServer/QuestDB 17 via `AggregateFamilies.Served`, Null **none**) is the one authority for both `AggregateFunctions` folders (i=2997, i=11201 — only the folder references are trimmed; the SDK's calculator factories stay for live `AggregateFilter` subscriptions) and the up-front `BadAggregateNotSupported` in HistoryRead-Processed. ⚠️ Historian disabled ⇒ Processed is now `BadAggregateNotSupported`, not `GoodNoData`. `docs/Historian.md` §Advertised aggregates. - **Real status fix:** the recorder used to stamp every sample Good (`192`); it now carries the driver's status. ⚠️ On the **Gateway** backend this changes nothing stored: the gateway discards quality (C-002), so AVEVA still stores Good. - **Comms loss is historized as an outage marker** (`Empty` + `BadNoCommunication`, one per tag per outage) on SqlServer; dropped (`DroppedUnsupportedKind`) on Gateway. Outbox records are **format v2** — drain or delete `ContinuousHistorization:OutboxPath` before downgrading a node. - **HistoryUpdate** (SqlServer + QuestDb; Gateway ⇒ `BadHistoryOperationUnsupported`): Insert/Replace/Update/DeleteRaw/DeleteAtTime, **Primary-only** since #674 (off-Primary ⇒ `BadInvalidState`, after authorization AND after planning — an unservable request (Remove / `IsDeleteModified` / not historized) stays `BadHistoryOperationUnsupported` off-Primary; the verdict is `IHistoryUpdateRoleView` = the data-plane verdict but **closed until this node's first redundancy snapshot**, so a pair cannot both accept in the boot window; admin-only nodes register no view ⇒ no gate; reads ungated), items with a `SourceTimestamp` past now + `ServerHistorian:HistoryUpdateMaxFutureSkew` (default 1 day, ≤0 disables) ⇒ `BadOutOfRange` in the node manager (every backend), gated in both ACL modes — flag off needs the `HistoryUpdate` role from a **Cluster-scoped** grant (+ key re-check), flag on uses `DataPlaneAuthorizer`'s HistoryUpdate operation; `WriteOperate` never implies it. The `HistoryWrite` AccessLevel bit is set only when an updater is wired at materialise time. `docs/Historian.md` §HistoryUpdate, `docs/security.md`. ⚠️ **SqlServer schema v1 is accepted (#674)** with `AutoCreateSchema=false`: writes/reads/events/retention work; modified reads + HistoryUpdate ⇒ node-level `BadHistoryOperationUnsupported` on every arm (the updater throws `NotSupportedException`, nothing sent; the `HistoryWrite` bit is still advertised — not knowable at materialise time), retention skips `SampleModified`, one Warning per boot (`SqlServerSchemaBootstrapper.SupportsModifiedTrail`; `RequiredVersion` = 1, `ModifiedTrailVersion` = 2). QuestDB: keep `cairo.ttl.use.wall.clock=true`; a QuestDB trail INSERT retried after a client timeout duplicates (no DEDUP key). @@ -1362,7 +1363,7 @@ historian tagname, which is the tag's driver `FullName`. The server dispatches all OPC UA HistoryRead to the registered `IHistorianDataSource` — the `GatewayHistorianDataSource` read client when enabled, else the `NullHistorianDataSource` default -(historized nodes return `GoodNoData`, never an error). Supported variants: Raw, Processed +(historized nodes return `GoodNoData` for Raw/AtTime — Processed is `BadAggregateNotSupported` since #682). Supported variants: Raw, Processed (backend-dependent: Gateway serves TimeAverage/Minimum/Maximum/Total, with Average/Count `BadAggregateNotSupported` since AVEVA has no mean or sample count, #679; SqlServer/QuestDb serve seventeen), and AtTime over historized variable nodes; Events over alarm-owning equipment-folder event-notifier nodes. Reads are ungated (served from any redundancy node); authorization uses the standard `AccessLevels.HistoryRead` bit set at materialization. diff --git a/docs/Historian.md b/docs/Historian.md index faa1b876..ed4eedb4 100644 --- a/docs/Historian.md +++ b/docs/Historian.md @@ -95,7 +95,9 @@ HistoryRead is served from any node (the database is shared). HistoryUpdate is * The `ServerHistorian` section in `appsettings.json` controls the historian read path. `Enabled` defaults to `false`; when disabled, the server registers `NullHistorianDataSource` -and all HistoryRead calls on historized nodes return `GoodNoData` (empty, not an error). +and Raw / AtTime / Events HistoryRead calls on historized nodes return `GoodNoData` (empty, not an error); a +Processed read is `BadAggregateNotSupported` and no aggregate functions are advertised +([#682](#advertised-aggregates-aggregatefunctions-682)). ```json { @@ -563,7 +565,7 @@ node without `isHistorized` has no HistoryRead access bit and answers |---|---| | Historized node, historian configured and reachable, results found | `Good` | | Historized node, historian configured, time range empty | `GoodNoData` | -| Historized node, historian NOT configured (`Enabled=false` / Null source) | `GoodNoData` (empty) | +| 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=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 | @@ -764,14 +766,46 @@ VPN. Reverse processed reads on the `Gateway` backend are answered by AVEVA and `HistoryAggregateType` carries the wider Part 13 set (Interpolative, TimeAverage, Start, End, Delta, Range, Min/MaxActualTime, Start/EndBound, DeltaBounds, DurationGood/Bad, PercentGood/Bad, WorstQuality, -NumberOfTransitions, AnnotationCount, StandardDeviationPopulation, VariancePopulation); `OtOpcUaNodeManager.MapAggregate` -maps each standard aggregate NodeId (`TimeAverage2` and `Total2` were appended for #671). The **`SqlServer`** and +NumberOfTransitions, AnnotationCount, StandardDeviationPopulation, VariancePopulation); `AggregateFunctionMap` +(`OpcUaServer`) maps each standard aggregate NodeId (`TimeAverage2` and `Total2` were appended for #671). The **`SqlServer`** and **`QuestDb`** backends serve the seventeen listed in the [aggregate table](#aggregates-processed-reads); everything else (e.g. `StartBound`, `DurationGood`, `StandardDeviationPopulation`) is `BadAggregateNotSupported` for that node. The **HistorianGateway** backend only has retrieval modes for the four -(TimeAverage, Minimum, Maximum, Total): for any other member `AggregateModeMapper` throws -`HistoryAggregateNotSupportedException`, which the HistoryRead path turns into `BadAggregateNotSupported` for that -node only (not a generic failure, and not counted as a backend fault). The OPC UA Client driver forwards every -member upstream. A NodeId outside the mapped set (e.g. `StandardDeviationSample`) is still `BadAggregateNotSupported`. +(TimeAverage, Minimum, Maximum, Total). A NodeId outside the mapped set (e.g. `StandardDeviationSample`) is still +`BadAggregateNotSupported`. + +### Advertised aggregates (`AggregateFunctions`, #682) + +Each backend declares the aggregates it serves as `IHistorianDataSource.SupportedAggregates` — **the single authority** for +two things: + +- **What the server advertises.** `Server/ServerCapabilities/AggregateFunctions` (`i=2997`) and + `HistoryServerCapabilities/AggregateFunctions` (`i=11201`) — two separate folder nodes that organize the same + aggregate-function objects — list exactly that set. Before #682 both listed the SDK's full 37 regardless of backend. +- **What HistoryRead-Processed refuses up front.** An aggregate outside the set is `BadAggregateNotSupported` for that node + without a backend call (the backend still throws `HistoryAggregateNotSupportedException` for one as defence in depth — + derived from the same table, never a second list). + +| `ServerHistorian:Backend` | Advertised | Declared by | +|---|---|---| +| `Gateway` | 4 — TimeAverage, Minimum, Maximum, Total | `AggregateModeMapper.Served` (derived from its mapping switch) | +| `SqlServer`, `QuestDb` | 17 — the [aggregate table](#aggregates-processed-reads) | `AggregateFamilies.Served` (Sql.Common; every member with a family) | +| historian disabled (Null source) | **none** | `NullHistorianDataSource` | + +⚠️ **Client-visible: with no historian configured a processed HistoryRead is now `BadAggregateNotSupported`, not +`GoodNoData`.** An aggregate is a statistic the server cannot compute without a history, and the empty folders now say so; +Raw / AtTime / Events reads on the Null source still return the empty `GoodNoData` result. + +How it is wired: `OtOpcUaSdkServer` keeps the SDK's `CreateAggregateManager` (all 37 calculator factories) and then trims +only the folder ⇄ aggregate `Organizes` references to the node manager's current source — the Null default at boot; the +Host's `SetHistorianDataSource` (the backend resolved once from `ServerHistorian:Backend`) re-syncs it right after start, +and every later assignment re-syncs both directions. ⚠️ **The calculator factories are deliberately NOT removed**: a +live-data MonitoredItem with an `AggregateFilter` is validated and computed by the SDK's `AggregateManager`, not the +historian, so live aggregate subscriptions keep working for every SDK aggregate, advertised or not (pinned over the wire +by `AdvertisedAggregateFunctionsWireTests`). The aggregate-function object nodes themselves stay readable. A boot log line +names the advertised set. + +Not changed: the `HistoryServerCapabilities` booleans (`AccessHistoryDataCapability`, `InsertDataCapability`, …) are still +whatever the SDK sets — #682 touches only the two folders (follow-up candidate). ### Known limitations diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/Historian/HistoryReadContracts.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/Historian/HistoryReadContracts.cs index cb98f97a..7cd25976 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/Historian/HistoryReadContracts.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/Historian/HistoryReadContracts.cs @@ -55,6 +55,18 @@ public enum HistoryAggregateType Total2, } +/// Shared sets for (#682). +public static class HistoryAggregateTypes +{ + /// Every defined — the interface default. + public static readonly IReadOnlySet All = + System.Collections.Frozen.FrozenSet.ToFrozenSet(Enum.GetValues()); + + /// No aggregates — a source with no history to aggregate (the Null historian). + public static readonly IReadOnlySet None = + System.Collections.Frozen.FrozenSet.Empty; +} + /// /// Thrown by a history backend when it cannot serve the requested . /// The Server HistoryRead path translates it to BadAggregateNotSupported for that node only. diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/Historian/IHistorianDataSource.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/Historian/IHistorianDataSource.cs index e7fcf2c6..803a4012 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/Historian/IHistorianDataSource.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/Historian/IHistorianDataSource.cs @@ -174,6 +174,23 @@ public interface IHistorianDataSource : IDisposable /// long TimestampResolutionTicks => 1; + /// + /// The Part 13 aggregates this backend serves for a processed HistoryRead (Gitea #682) — the ONE authority for + /// both what the server advertises under ServerCapabilities/AggregateFunctions (i=2997) and + /// HistoryServerCapabilities/AggregateFunctions (i=11201), and what its HistoryRead-Processed answers + /// BadAggregateNotSupported up front without calling the backend. + /// + /// + /// The default is — "every aggregate the server can name; a member I cannot + /// serve throws at read time", which is exactly the pre-#682 + /// behaviour, so a test fake or an out-of-tree source keeps working unchanged. It is a concrete set rather than a + /// null "all" sentinel so every consumer handles one shape. Every PRODUCTION backend declares its set itself + /// (a reflection guard pins that): a wrapper that forgot to forward this would otherwise advertise every aggregate. + /// The set is read once when the source is wired (the backend is fixed at boot by ServerHistorian:Backend) + /// and again on every processed read, so it must be cheap and stable — return a static frozen set. + /// + IReadOnlySet SupportedAggregates => HistoryAggregateTypes.All; + /// /// Read modified-value history — OPC UA HistoryReadRaw with IsReadModified = true (Part 11, Gitea #668): /// every modification a HistoryUpdate applied to the tag inside the window, each carrying the PRIOR value. diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/Historian/NullHistorianDataSource.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/Historian/NullHistorianDataSource.cs index c5b3e4b2..d0553988 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/Historian/NullHistorianDataSource.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/Historian/NullHistorianDataSource.cs @@ -74,6 +74,15 @@ public sealed class NullHistorianDataSource : IHistorianDataSource ? throw new NotSupportedException("The null historian cannot read events newest-first (reverse).") : Task.FromResult(EmptyEvents); + /// + /// + /// EMPTY (Gitea #682): with no historian configured the server advertises no aggregate functions, and a processed + /// HistoryRead answers BadAggregateNotSupported per node rather than GoodNoData — an aggregate is a + /// statistic the server cannot compute without a history, and advertising the SDK's full list on a server with no + /// historian misled capability discovery. Raw / AtTime / Events reads still return the empty GoodNoData result. + /// + public IReadOnlySet SupportedAggregates => HistoryAggregateTypes.None; + /// public HistorianHealthSnapshot GetHealthSnapshot() => new( TotalQueries: 0, 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 34e25ad2..c8ac094a 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 @@ -132,6 +132,13 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis /// public bool SupportsReverseEvents => false; + /// + /// + /// — the aggregates the gateway has a retrieval mode for (#682); every + /// other member answers BadAggregateNotSupported and is not advertised. + /// + public IReadOnlySet SupportedAggregates => AggregateModeMapper.Served; + /// /// /// A reverse (newest-first) read is refused with — the server reports diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/Mapping/AggregateModeMapper.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/Mapping/AggregateModeMapper.cs index 35654d2a..16ba239a 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/Mapping/AggregateModeMapper.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/Mapping/AggregateModeMapper.cs @@ -1,3 +1,4 @@ +using System.Collections.Frozen; using ZB.MOM.WW.HistorianGateway.Contracts.Grpc; using ZB.MOM.WW.OtOpcUa.Core.Abstractions; @@ -22,22 +23,38 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Mapping; /// internal static class AggregateModeMapper { + /// + /// The aggregates the gateway serves — every defined member maps, derived from it so the + /// advertised set and the dispatch can never disagree. + /// returns it (#682); it is the ONLY declaration of the gateway's served set. + /// + public static readonly IReadOnlySet Served = + Enum.GetValues().Where(a => TryMap(a) is not null).ToFrozenSet(); + /// Maps an aggregate function to the gateway retrieval mode. /// The driver-agnostic aggregate function. /// The matching gateway . /// A defined member the gateway has no retrieval mode for. /// An undefined enum value. - public static RetrievalMode ToRetrievalMode(HistoryAggregateType aggregate) => aggregate switch + public static RetrievalMode ToRetrievalMode(HistoryAggregateType aggregate) + { + if (TryMap(aggregate) is { } mode) return mode; + + // Part 13 aggregates the gateway has no retrieval mode for (#285) — an UNDEFINED enum value is a + // programming error, but a defined-yet-unserved member is a normal client request the server + // answers with BadAggregateNotSupported. + if (Enum.IsDefined(aggregate)) throw new HistoryAggregateNotSupportedException(aggregate); + throw new ArgumentOutOfRangeException( + nameof(aggregate), aggregate, "Unmapped HistoryAggregateType — add a RetrievalMode mapping."); + } + + /// The single mapping table: the retrieval mode for a served aggregate, or null. + private static RetrievalMode? TryMap(HistoryAggregateType aggregate) => aggregate switch { HistoryAggregateType.TimeAverage => RetrievalMode.TimeWeightedAverage, HistoryAggregateType.Minimum => RetrievalMode.MinimumWithTime, HistoryAggregateType.Maximum => RetrievalMode.MaximumWithTime, HistoryAggregateType.Total => RetrievalMode.Integral, - // Part 13 aggregates the gateway has no retrieval mode for (#285) — an UNDEFINED enum value is a - // programming error, but a defined-yet-unserved member is a normal client request the server - // answers with BadAggregateNotSupported. - _ when Enum.IsDefined(aggregate) => throw new HistoryAggregateNotSupportedException(aggregate), - _ => throw new ArgumentOutOfRangeException( - nameof(aggregate), aggregate, "Unmapped HistoryAggregateType — add a RetrievalMode mapping."), + _ => null, }; } 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 c09e6361..f192a2b5 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 @@ -359,6 +359,10 @@ public sealed class QuestDbHistorianDataSource : IHistorianDataSource /// Reverse events reads are served natively (ORDER BY … DESC, Gitea #675). public bool SupportsReverseEvents => true; + /// + /// The seventeen — shared with the other SQL backend (#682). + public IReadOnlySet SupportedAggregates => AggregateFamilies.Served; + /// /// diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common/AggregateFamily.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common/AggregateFamily.cs index 18066b09..575662c6 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common/AggregateFamily.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common/AggregateFamily.cs @@ -1,3 +1,4 @@ +using System.Collections.Frozen; using ZB.MOM.WW.OtOpcUa.Core.Abstractions; namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common; @@ -40,4 +41,12 @@ public static class AggregateFamilies or HistoryAggregateType.PercentGood or HistoryAggregateType.PercentBad => AggregateFamily.TimeWeighted, _ => null, }; + + /// + /// The seventeen aggregates the SQL backends serve — every member assigns a family, derived + /// from it so the set and the dispatch can never disagree. Both SqlServerHistorianDataSource and + /// QuestDbHistorianDataSource return it as (#682). + /// + public static readonly IReadOnlySet Served = + Enum.GetValues().Where(a => FamilyOf(a) is not null).ToFrozenSet(); } 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 acef70f7..e1e458b0 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 @@ -316,6 +316,10 @@ public sealed class SqlServerHistorianDataSource : IHistorianDataSource /// Reverse events reads are served natively (ORDER BY … DESC, Gitea #675). public bool SupportsReverseEvents => true; + /// + /// The seventeen — shared with the other SQL backend (#682). + public IReadOnlySet SupportedAggregates => AggregateFamilies.Served; + /// /// diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/AdvertisedAggregateFunctions.cs b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/AdvertisedAggregateFunctions.cs new file mode 100644 index 00000000..dd475de4 --- /dev/null +++ b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/AdvertisedAggregateFunctions.cs @@ -0,0 +1,91 @@ +using System.Reflection; +using Opc.Ua; +using Opc.Ua.Server; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; + +namespace ZB.MOM.WW.OtOpcUa.OpcUaServer; + +/// +/// Gitea #682 — make the two advertised aggregate-function folders, ServerCapabilities/AggregateFunctions +/// (i=2997) and HistoryServerCapabilities/AggregateFunctions (i=11201), list exactly the aggregates the wired +/// historian backend serves (), instead of the SDK's full +/// registered set. +/// +/// +/// +/// The SDK's AggregateManager.RegisterFactory both registers a calculator factory AND adds an +/// Organizes reference from each folder to the aggregate's object node. Only the REFERENCES are trimmed +/// here: the factories stay, because a live-data MonitoredItem with an AggregateFilter is validated +/// against (and computed by) the AggregateManager, not the historian — removing a factory would break live +/// aggregate subscriptions for every aggregate the historian happens not to serve. +/// +/// +/// The two folders are separate nodes that reference the same aggregate object nodes (both directions). The +/// aggregate nodes themselves are kept (they are standard type-system nodes a client may still Read); only the +/// folder ⇄ aggregate Organizes pair is removed or restored. Idempotent and reversible, so re-wiring a +/// different source re-syncs both folders. +/// +/// +internal static class AdvertisedAggregateFunctions +{ + /// Every standard aggregate-function NodeId the SDK's nodeset knows (ObjectIds.AggregateFunction_*). + private static readonly NodeId[] StandardAggregateIds = typeof(ObjectIds) + .GetFields(BindingFlags.Public | BindingFlags.Static) + .Where(f => f.FieldType == typeof(NodeId) && f.Name.StartsWith("AggregateFunction_", StringComparison.Ordinal)) + .Select(f => (NodeId)f.GetValue(null)!) + .ToArray(); + + /// The two advertised folders (both are trimmed — they are separate nodes). + internal static readonly NodeId[] Folders = + [ + ObjectIds.Server_ServerCapabilities_AggregateFunctions, + ObjectIds.HistoryServerCapabilities_AggregateFunctions, + ]; + + /// + /// Sync both folders to : an aggregate-function node is organized under a folder iff the + /// server maps it () AND contains it. + /// + /// The running server. + /// The backend's served aggregates. + /// The advertised NodeIds, or null when the diagnostics node manager is not available yet. + public static IReadOnlyList? Apply(IServerInternal server, IReadOnlySet served) + { + if (server?.DiagnosticsNodeManager is not CustomNodeManager2 diagnostics) return null; + + var advertised = new List(); + lock (diagnostics.Lock) + { + var folders = Folders + .Select(id => diagnostics.FindPredefinedNode(id, typeof(BaseObjectState))) + .Where(f => f is not null) + .ToList(); + + foreach (var aggregateId in StandardAggregateIds) + { + var aggregateNode = diagnostics.FindPredefinedNode(aggregateId, null); + if (aggregateNode is null) continue; + + var advertise = AggregateFunctionMap.Map(aggregateId) is { } aggregate && served.Contains(aggregate); + if (advertise) advertised.Add(aggregateId); + + foreach (var folder in folders) + { + var linked = folder!.ReferenceExists(ReferenceTypeIds.Organizes, false, aggregateId); + if (advertise && !linked) + { + folder.AddReference(ReferenceTypeIds.Organizes, false, aggregateId); + aggregateNode.AddReference(ReferenceTypeIds.Organizes, true, folder.NodeId); + } + else if (!advertise && linked) + { + folder.RemoveReference(ReferenceTypeIds.Organizes, false, aggregateId); + aggregateNode.RemoveReference(ReferenceTypeIds.Organizes, true, folder.NodeId); + } + } + } + } + + return advertised; + } +} diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/AggregateFunctionMap.cs b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/AggregateFunctionMap.cs new file mode 100644 index 00000000..8ade15ea --- /dev/null +++ b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/AggregateFunctionMap.cs @@ -0,0 +1,55 @@ +using System.Collections.Frozen; +using Opc.Ua; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; + +namespace ZB.MOM.WW.OtOpcUa.OpcUaServer; + +/// +/// The one table between OPC UA Part 13 standard aggregate-function NodeIds (ObjectIds.AggregateFunction_*) +/// and . HistoryRead-Processed maps a request's aggregate through it, and the +/// advertised AggregateFunctions folders (#682) map a backend's served set back through it — one table, so a +/// NodeId the server can dispatch is exactly a NodeId it can advertise. +/// +internal static class AggregateFunctionMap +{ + /// NodeId ⇒ aggregate, for every aggregate the server can name to a backend. + public static readonly FrozenDictionary ByNodeId = new Dictionary + { + [ObjectIds.AggregateFunction_Average] = HistoryAggregateType.Average, + [ObjectIds.AggregateFunction_Minimum] = HistoryAggregateType.Minimum, + [ObjectIds.AggregateFunction_Maximum] = HistoryAggregateType.Maximum, + [ObjectIds.AggregateFunction_Total] = HistoryAggregateType.Total, + [ObjectIds.AggregateFunction_Count] = HistoryAggregateType.Count, + [ObjectIds.AggregateFunction_Interpolative] = HistoryAggregateType.Interpolative, + [ObjectIds.AggregateFunction_TimeAverage] = HistoryAggregateType.TimeAverage, + [ObjectIds.AggregateFunction_Start] = HistoryAggregateType.Start, + [ObjectIds.AggregateFunction_End] = HistoryAggregateType.End, + [ObjectIds.AggregateFunction_Delta] = HistoryAggregateType.Delta, + [ObjectIds.AggregateFunction_Range] = HistoryAggregateType.Range, + [ObjectIds.AggregateFunction_MinimumActualTime] = HistoryAggregateType.MinimumActualTime, + [ObjectIds.AggregateFunction_MaximumActualTime] = HistoryAggregateType.MaximumActualTime, + [ObjectIds.AggregateFunction_StartBound] = HistoryAggregateType.StartBound, + [ObjectIds.AggregateFunction_EndBound] = HistoryAggregateType.EndBound, + [ObjectIds.AggregateFunction_DeltaBounds] = HistoryAggregateType.DeltaBounds, + [ObjectIds.AggregateFunction_DurationGood] = HistoryAggregateType.DurationGood, + [ObjectIds.AggregateFunction_DurationBad] = HistoryAggregateType.DurationBad, + [ObjectIds.AggregateFunction_PercentGood] = HistoryAggregateType.PercentGood, + [ObjectIds.AggregateFunction_PercentBad] = HistoryAggregateType.PercentBad, + [ObjectIds.AggregateFunction_WorstQuality] = HistoryAggregateType.WorstQuality, + [ObjectIds.AggregateFunction_NumberOfTransitions] = HistoryAggregateType.NumberOfTransitions, + [ObjectIds.AggregateFunction_AnnotationCount] = HistoryAggregateType.AnnotationCount, + [ObjectIds.AggregateFunction_StandardDeviationPopulation] = HistoryAggregateType.StandardDeviationPopulation, + [ObjectIds.AggregateFunction_VariancePopulation] = HistoryAggregateType.VariancePopulation, + [ObjectIds.AggregateFunction_TimeAverage2] = HistoryAggregateType.TimeAverage2, + [ObjectIds.AggregateFunction_Total2] = HistoryAggregateType.Total2, + }.ToFrozenDictionary(); + + /// + /// Map a request's aggregate-function NodeId to ; null for one the server + /// cannot name to any backend (the caller answers BadAggregateNotSupported). + /// + /// The per-node aggregate-function NodeId from the request. + /// The mapped aggregate, or null. + public static HistoryAggregateType? Map(NodeId? aggregateNodeId) => + aggregateNodeId is not null && ByNodeId.TryGetValue(aggregateNodeId, out var aggregate) ? aggregate : null; +} diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs index 548dbcb4..eba945fc 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs @@ -344,11 +344,48 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 /// , the HistoryRead override does NOT run under the /// node-manager Lock, so the override may block-bridge to this (async) source. /// + /// + /// Gitea #682: assigning a source also re-syncs the advertised AggregateFunctions folders to its + /// (see ). + /// /// public IHistorianDataSource HistorianDataSource { get => _historianDataSource; - set => _historianDataSource = value ?? NullHistorianDataSource.Instance; + set + { + _historianDataSource = value ?? NullHistorianDataSource.Instance; + SyncAdvertisedAggregates(); + } + } + + /// + /// Gitea #682 — trim ServerCapabilities/AggregateFunctions (i=2997) and + /// HistoryServerCapabilities/AggregateFunctions (i=11201) to the current source's + /// . Called on every + /// assignment and by right after the SDK registers its aggregate factories, so the + /// Null default is applied from boot. Only the folder references change — the AggregateManager's calculator + /// factories (live AggregateFilter subscriptions) are untouched. Never throws: a failure leaves the previous + /// advertisement in place and is logged, because HistoryRead's own up-front check still enforces the set. + /// + internal void SyncAdvertisedAggregates() + { + try + { + var source = _historianDataSource; + var advertised = AdvertisedAggregateFunctions.Apply(Server, source.SupportedAggregates); + if (advertised is not null) + { + _logger.LogInformation( + "OtOpcUaNodeManager: advertising {Count} aggregate function(s) for historian {Source}: {Aggregates}", + advertised.Count, source.GetType().Name, + string.Join(", ", advertised.Select(id => AggregateFunctionMap.Map(id)?.ToString() ?? id.ToString()))); + } + } + catch (Exception ex) + { + _logger.LogWarning(ex, "OtOpcUaNodeManager: failed to sync the advertised aggregate functions"); + } } private volatile IHistorianDataUpdater? _historianDataUpdater; @@ -5054,6 +5091,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 // #677: a client AggregateConfiguration other than the server defaults is refused per node, never silently ignored. var configurationServable = ServerAggregateConfiguration.IsServable(details.AggregateConfiguration); nodesToProcess = AuthorizeHistoryRead(context, nodesToProcess, results, errors); + var source = HistorianDataSource; WithHistoryReadBatchLimiter(nodesToProcess, results, errors, () => { using var deadline = CreateDeadline(); @@ -5064,8 +5102,11 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 // AggregateType is a per-node parallel collection (same length as nodesToRead, enforced by // the base dispatcher). handle.Index is the node's position in that collection. var aggregateNodeId = details.AggregateType[handle.Index]; - var aggregate = MapAggregate(aggregateNodeId); - if (aggregate is null) + // #682: one authority — the source's SupportedAggregates (also what the AggregateFunctions folders + // advertise). An aggregate the server cannot name, or the backend does not serve, is refused up front + // without a backend call. + var aggregate = AggregateFunctionMap.Map(aggregateNodeId); + if (aggregate is null || !source.SupportedAggregates.Contains(aggregate.Value)) { errors[handle.Index] = StatusCodes.BadAggregateNotSupported; continue; @@ -6478,45 +6519,6 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 return new HistoryData { DataValues = values }; } - /// - /// Map an OPC UA Part 13 standard-aggregate function NodeId to our - /// . Returns null for any aggregate we don't serve so - /// the caller can surface BadAggregateNotSupported. - /// - /// The per-node aggregate-function NodeId from the request. - /// The mapped aggregate, or null when unsupported. - private static HistoryAggregateType? MapAggregate(NodeId aggregateNodeId) - { - if (aggregateNodeId == ObjectIds.AggregateFunction_Average) return HistoryAggregateType.Average; - if (aggregateNodeId == ObjectIds.AggregateFunction_Minimum) return HistoryAggregateType.Minimum; - if (aggregateNodeId == ObjectIds.AggregateFunction_Maximum) return HistoryAggregateType.Maximum; - if (aggregateNodeId == ObjectIds.AggregateFunction_Total) return HistoryAggregateType.Total; - if (aggregateNodeId == ObjectIds.AggregateFunction_Count) return HistoryAggregateType.Count; - if (aggregateNodeId == ObjectIds.AggregateFunction_Interpolative) return HistoryAggregateType.Interpolative; - if (aggregateNodeId == ObjectIds.AggregateFunction_TimeAverage) return HistoryAggregateType.TimeAverage; - if (aggregateNodeId == ObjectIds.AggregateFunction_Start) return HistoryAggregateType.Start; - if (aggregateNodeId == ObjectIds.AggregateFunction_End) return HistoryAggregateType.End; - if (aggregateNodeId == ObjectIds.AggregateFunction_Delta) return HistoryAggregateType.Delta; - if (aggregateNodeId == ObjectIds.AggregateFunction_Range) return HistoryAggregateType.Range; - if (aggregateNodeId == ObjectIds.AggregateFunction_MinimumActualTime) return HistoryAggregateType.MinimumActualTime; - if (aggregateNodeId == ObjectIds.AggregateFunction_MaximumActualTime) return HistoryAggregateType.MaximumActualTime; - if (aggregateNodeId == ObjectIds.AggregateFunction_StartBound) return HistoryAggregateType.StartBound; - if (aggregateNodeId == ObjectIds.AggregateFunction_EndBound) return HistoryAggregateType.EndBound; - if (aggregateNodeId == ObjectIds.AggregateFunction_DeltaBounds) return HistoryAggregateType.DeltaBounds; - if (aggregateNodeId == ObjectIds.AggregateFunction_DurationGood) return HistoryAggregateType.DurationGood; - if (aggregateNodeId == ObjectIds.AggregateFunction_DurationBad) return HistoryAggregateType.DurationBad; - if (aggregateNodeId == ObjectIds.AggregateFunction_PercentGood) return HistoryAggregateType.PercentGood; - if (aggregateNodeId == ObjectIds.AggregateFunction_PercentBad) return HistoryAggregateType.PercentBad; - if (aggregateNodeId == ObjectIds.AggregateFunction_WorstQuality) return HistoryAggregateType.WorstQuality; - if (aggregateNodeId == ObjectIds.AggregateFunction_NumberOfTransitions) return HistoryAggregateType.NumberOfTransitions; - if (aggregateNodeId == ObjectIds.AggregateFunction_AnnotationCount) return HistoryAggregateType.AnnotationCount; - if (aggregateNodeId == ObjectIds.AggregateFunction_StandardDeviationPopulation) return HistoryAggregateType.StandardDeviationPopulation; - if (aggregateNodeId == ObjectIds.AggregateFunction_VariancePopulation) return HistoryAggregateType.VariancePopulation; - if (aggregateNodeId == ObjectIds.AggregateFunction_TimeAverage2) return HistoryAggregateType.TimeAverage2; - if (aggregateNodeId == ObjectIds.AggregateFunction_Total2) return HistoryAggregateType.Total2; - return null; - } - /// /// Project the historian source's (Core.Abstractions DTO) into an /// SDK — one per , diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaSdkServer.cs b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaSdkServer.cs index e566b347..67af4072 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaSdkServer.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaSdkServer.cs @@ -200,6 +200,21 @@ public sealed class OtOpcUaSdkServer : StandardServer return true; } + /// + /// Gitea #682 — keep the SDK's full set of calculator factories (live-data AggregateFilter subscriptions + /// need them), then trim the advertised AggregateFunctions folders to the node manager's current historian + /// source. At this point that is the Null default (no aggregates); the Host's + /// re-syncs to the configured backend's set right after start. + /// + /// + protected override AggregateManager CreateAggregateManager( + IServerInternal server, ApplicationConfiguration configuration) + { + var manager = base.CreateAggregateManager(server, configuration); + _otOpcUaNodeManager?.SyncAdvertisedAggregates(); + return manager; + } + /// protected override MasterNodeManager CreateMasterNodeManager( IServerInternal server, ApplicationConfiguration configuration) diff --git a/tests/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions.Tests/Historian/HistorianDataSourceReverseEventsGuardTests.cs b/tests/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions.Tests/Historian/HistorianDataSourceReverseEventsGuardTests.cs index 520ba3b0..fdc4d365 100644 --- a/tests/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions.Tests/Historian/HistorianDataSourceReverseEventsGuardTests.cs +++ b/tests/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions.Tests/Historian/HistorianDataSourceReverseEventsGuardTests.cs @@ -63,4 +63,19 @@ public sealed class HistorianDataSourceReverseEventsGuardTests $"{type.FullName} must declare SupportsReverseEvents itself, not inherit the interface default"); } } + + [Fact] + public void Every_implementation_declares_its_supported_aggregates_itself() + { + // Gitea #682: the interface default is "every aggregate" — a production backend (or a future wrapper) that + // inherited it would advertise all of them under AggregateFunctions and forward reads it cannot serve. + foreach (var type in ProductionImplementations()) + { + var map = type.GetInterfaceMap(typeof(IHistorianDataSource)); + var getter = map.TargetMethods[Array.FindIndex(map.InterfaceMethods, + m => m.Name == "get_" + nameof(IHistorianDataSource.SupportedAggregates))]; + getter.DeclaringType.ShouldBe(type, + $"{type.FullName} must declare SupportedAggregates itself, not inherit the interface default"); + } + } } diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/Mapping/AggregateModeMapperTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/Mapping/AggregateModeMapperTests.cs index 8a8c0857..a468a297 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/Mapping/AggregateModeMapperTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/Mapping/AggregateModeMapperTests.cs @@ -38,6 +38,21 @@ public sealed class AggregateModeMapperTests public void Average_and_Count_are_not_supported(HistoryAggregateType a) => Assert.Throws(() => AggregateModeMapper.ToRetrievalMode(a)); + [Fact] // #682: the served (= advertised) set is exactly what the mapper maps, and the data source returns it + public void Served_is_exactly_the_mapped_members_and_the_data_source_declares_it() + { + Assert.Equal( + new[] { HistoryAggregateType.Minimum, HistoryAggregateType.Maximum, HistoryAggregateType.Total, HistoryAggregateType.TimeAverage } + .OrderBy(a => a), + AggregateModeMapper.Served.OrderBy(a => a)); + foreach (var a in Enum.GetValues().Where(a => !AggregateModeMapper.Served.Contains(a))) + Assert.Throws(() => AggregateModeMapper.ToRetrievalMode(a)); + + var source = new GatewayHistorianDataSource( + new FakeHistorianGatewayClient(), Microsoft.Extensions.Logging.Abstractions.NullLogger.Instance); + Assert.Same(AggregateModeMapper.Served, source.SupportedAggregates); + } + [Fact] public void Undefined_value_is_a_programming_error_not_a_not_supported() => Assert.Throws(() => AggregateModeMapper.ToRetrievalMode((HistoryAggregateType)9999)); diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common.Tests/ProcessedReadTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common.Tests/ProcessedReadTests.cs index 461c0b82..a63e46d0 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common.Tests/ProcessedReadTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common.Tests/ProcessedReadTests.cs @@ -19,4 +19,13 @@ public sealed class ProcessedReadTests HistoryAggregateType.PercentGood, HistoryAggregateType.PercentBad, HistoryAggregateType.TimeAverage2, HistoryAggregateType.Total2]); } + + [Fact] + public void Served_is_exactly_the_members_with_a_family() + { + // #682: the advertised set and the dispatch are one definition — a member is served iff it has a family. + AggregateFamilies.Served.Count.ShouldBe(17); + foreach (var a in Enum.GetValues()) + AggregateFamilies.Served.Contains(a).ShouldBe(AggregateFamilies.FamilyOf(a) is not null, a.ToString()); + } } diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Host.Tests/Configuration/HistorianBackendRegistryTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.Tests/Configuration/HistorianBackendRegistryTests.cs index fe66467c..bcc033ba 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Host.Tests/Configuration/HistorianBackendRegistryTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.Tests/Configuration/HistorianBackendRegistryTests.cs @@ -129,4 +129,33 @@ public sealed class HistorianBackendRegistryTests o.ConnectionString.ShouldContain("Port=18812"); o.TablePrefix.ShouldBe("rig_"); } + + /// 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). + 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(); + } } diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/AdvertisedAggregateFunctionsWireTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/AdvertisedAggregateFunctionsWireTests.cs new file mode 100644 index 00000000..8c7cb4f6 --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/AdvertisedAggregateFunctionsWireTests.cs @@ -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; + +/// +/// Gitea #682 over a real SDK client session: ServerCapabilities/AggregateFunctions (i=2997) and +/// HistoryServerCapabilities/AggregateFunctions (i=11201) list exactly the aggregates the wired historian +/// backend serves — the data sources are built through exactly as the Host +/// builds them — while a live-data AggregateFilter subscription keeps working for every aggregate the SDK +/// calculates, advertised or not. +/// +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 + { + ["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(); } + } + + /// Forward Organizes children of an AggregateFunctions folder, by BrowseName, sorted. + private static async Task> 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(); + } + + /// The folders an aggregate-function node is organized under (inverse Organizes). + private static async Task> 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 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; + } +} 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 d2b0110f..d119e496 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadTests.cs @@ -278,6 +278,54 @@ public sealed class NodeManagerHistoryReadTests : IDisposable await host.DisposeAsync(); } + /// + /// #682: an aggregate the server maps but the wired backend does not declare in + /// is refused BadAggregateNotSupported 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. + /// + [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.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(); + } + /// /// #677: a client AggregateConfiguration that differs from the server defaults is refused per node with /// BadAggregateConfigurationRejected (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 Supported { get; init; } = HistoryAggregateTypes.All; + public IReadOnlySet SupportedAggregates => Supported; public HistorianRead RawResult { get; set; } = new(Array.Empty(), null); public HistorianRead ProcessedResult { get; set; } = new(Array.Empty(), null); public HistorianRead AtTimeResult { get; set; } = new(Array.Empty(), null); From ede86108048d0c3f87d085fd2200bb11ed4198ac Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Mon, 12 Oct 2026 08:57:19 -0400 Subject: [PATCH 2/3] refactor(historian): Gateway served set behind GatewayHistorianDataSource.ServedAggregates (#682, merge seam for #681) Co-Authored-By: Claude --- CLAUDE.md | 2 +- docs/Historian.md | 2 +- .../GatewayHistorianDataSource.cs | 14 +++++++++----- .../Mapping/AggregateModeMapperTests.cs | 3 ++- 4 files changed, 13 insertions(+), 8 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index 08a6f7a0..6c1f9d0d 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -1353,7 +1353,7 @@ historian tagname, which is the tag's driver `FullName`. - **SQL Server** (`Driver.Historian.SqlServer`): `hist.Sample` PK `(TagId, SourceTimeUtc)` — writes are insert-if-absent, so outbox redelivery is idempotent. Stores every value kind with its **real status**; per-sample aggregates (10, simple definitions) are pushed down to T-SQL with bucket math in **100 ns ticks**; the Part 13 time-weighted seven (TimeAverage/TimeAverage2/Total/Total2/Interpolative, #671 — ⚠️ `Total` is now the time-weighted integral, no longer `SUM` — plus the **duration-based** PercentGood/PercentBad, #680) stream window + 4 bound rows into the shared `TimeWeightedAggregator` (both SQL backends, identical results; reverse `Interpolative` anchored at the later time; window capped by `ServerHistorian:MaxTimeWeightedRawRows` ⇒ `BadTooManyOperations`, percentages included). ⚠️ **#680 status rules (client-visible):** TimeAverage/Total are Uncertain if any part of the interval uses an Uncertain value, is bridged across a Bad sample, is extrapolated past the last sample (⇒ a live tag's latest interval is Uncertain) or is not covered; TimeAverage2/Total2 keep the SDK time-based rule (Uncertain time counts as Good) but never extrapolate past the last sample — that interval is Partial, later ones `BadNoData` (a report-by-exception tag with nothing after its last sample reads `BadNoData` until its next sample arrives). `docs/Historian.md` §"Time-weighted aggregates". ⚠️ **#680 per-sample (client-visible):** `Count` counts **Good** samples only (value-based status; an all-Bad interval is generic Bad, value null) and an empty interval **inside the tag's data range** is `0`/Good (outside it still `BadNoData`); `Start`/`End` return the **raw** first/last sample of any quality — its own timestamp and status, value null when Bad; `Delta` is Uncertain only when the interval's first or last sample is unusable; and every per-sample aggregate but `Average` sets **end-of-data Partial** on the interval holding the last sample (forward incl. its earlier edge, reverse strictly inside), never on a Bad result. The data-range lookup is a seek on SQL Server but a partition scan on QuestDB (unindexed `tag` SYMBOL). `docs/Historian.md` §"Aggregates". Docker-dev recipe, schema and aggregate semantics: `docs/Historian.md` §"SQL Server backend". Integration tests need `HISTORIAN_SQL_TEST_CONNECTION`. **Retention (#667):** opt-in `SqlServer:RetentionDays` (+ `PurgeInterval` 1h, `PurgeBatchSize` 4000) -> `HistorianRetentionService` purges old `hist.Sample` (per tag) / `hist.Event` in `DELETE TOP(n)` batches, Primary-gated on `IHistorizationRoleView`, one purger per DB via `sp_getapplock`; counter `otopcua.historian.retention.purged{table}`. Optional `IHistorianBackendProvider.CreateRetention` (default null). `docs/Historian.md` §Retention. - **QuestDB** (`Driver.Historian.QuestDb`, `ServerHistorian:QuestDb:*`, Npgsql over PG wire; docker-dev overlay `docker-compose.questdb.yml`): same kinds/status/aggregates/HistoryUpdate as SqlServer, soft delete by tombstone (INSERT only — never `UPDATE`/`DELETE`; every read filters `NOT deleted`). ⚠️ **Timestamps are µs** (`QuestDbConnection.Round`/`BindTimestamp`; `IHistorianDataSource.TimestampResolutionTicks` = 10 steps HistoryRead paging), **pre-1970 cannot be stored** (writer drops, HistoryUpdate `BadOutOfRange`), every write waits for WAL visibility, TTL is enforced on write commits only, set Npgsql `Maximum Pool Size` < QuestDB's 64. `docs/Historian.md` §"QuestDB backend". - ⚠️ **Reverse processed reads are Part 13/SDK-anchored (#677, client-visible):** every aggregate on SqlServer/QuestDB now anchors a reverse read at its later time — window `(end, start]`, partial interval earliest, each stamped with its later edge (was: earlier-anchored, stamped earlier edge); on forward reads Partial also marks the interval the data starts inside (not Average — the SDK's rule, forward-only); a non-default client `AggregateConfiguration` ⇒ `BadAggregateConfigurationRejected`; Gateway provisioning sends `IntegralDivisor = 1`. SDK-parity tests: `SdkAggregateParityTests`. `docs/Historian.md` §Aggregates. -- **Advertised aggregates (#682):** `IHistorianDataSource.SupportedAggregates` (Gateway 4 via `AggregateModeMapper.Served`, SqlServer/QuestDB 17 via `AggregateFamilies.Served`, Null **none**) is the one authority for both `AggregateFunctions` folders (i=2997, i=11201 — only the folder references are trimmed; the SDK's calculator factories stay for live `AggregateFilter` subscriptions) and the up-front `BadAggregateNotSupported` in HistoryRead-Processed. ⚠️ Historian disabled ⇒ Processed is now `BadAggregateNotSupported`, not `GoodNoData`. `docs/Historian.md` §Advertised aggregates. +- **Advertised aggregates (#682):** `IHistorianDataSource.SupportedAggregates` (Gateway 4 via `GatewayHistorianDataSource.ServedAggregates`, SqlServer/QuestDB 17 via `AggregateFamilies.Served`, Null **none**) is the one authority for both `AggregateFunctions` folders (i=2997, i=11201 — only the folder references are trimmed; the SDK's calculator factories stay for live `AggregateFilter` subscriptions) and the up-front `BadAggregateNotSupported` in HistoryRead-Processed. ⚠️ Historian disabled ⇒ Processed is now `BadAggregateNotSupported`, not `GoodNoData`. `docs/Historian.md` §Advertised aggregates. - **Real status fix:** the recorder used to stamp every sample Good (`192`); it now carries the driver's status. ⚠️ On the **Gateway** backend this changes nothing stored: the gateway discards quality (C-002), so AVEVA still stores Good. - **Comms loss is historized as an outage marker** (`Empty` + `BadNoCommunication`, one per tag per outage) on SqlServer; dropped (`DroppedUnsupportedKind`) on Gateway. Outbox records are **format v2** — drain or delete `ContinuousHistorization:OutboxPath` before downgrading a node. - **HistoryUpdate** (SqlServer + QuestDb; Gateway ⇒ `BadHistoryOperationUnsupported`): Insert/Replace/Update/DeleteRaw/DeleteAtTime, **Primary-only** since #674 (off-Primary ⇒ `BadInvalidState`, after authorization AND after planning — an unservable request (Remove / `IsDeleteModified` / not historized) stays `BadHistoryOperationUnsupported` off-Primary; the verdict is `IHistoryUpdateRoleView` = the data-plane verdict but **closed until this node's first redundancy snapshot**, so a pair cannot both accept in the boot window; admin-only nodes register no view ⇒ no gate; reads ungated), items with a `SourceTimestamp` past now + `ServerHistorian:HistoryUpdateMaxFutureSkew` (default 1 day, ≤0 disables) ⇒ `BadOutOfRange` in the node manager (every backend), gated in both ACL modes — flag off needs the `HistoryUpdate` role from a **Cluster-scoped** grant (+ key re-check), flag on uses `DataPlaneAuthorizer`'s HistoryUpdate operation; `WriteOperate` never implies it. The `HistoryWrite` AccessLevel bit is set only when an updater is wired at materialise time. `docs/Historian.md` §HistoryUpdate, `docs/security.md`. ⚠️ **SqlServer schema v1 is accepted (#674)** with `AutoCreateSchema=false`: writes/reads/events/retention work; modified reads + HistoryUpdate ⇒ node-level `BadHistoryOperationUnsupported` on every arm (the updater throws `NotSupportedException`, nothing sent; the `HistoryWrite` bit is still advertised — not knowable at materialise time), retention skips `SampleModified`, one Warning per boot (`SqlServerSchemaBootstrapper.SupportsModifiedTrail`; `RequiredVersion` = 1, `ModifiedTrailVersion` = 2). QuestDB: keep `cairo.ttl.use.wall.clock=true`; a QuestDB trail INSERT retried after a client timeout duplicates (no DEDUP key). diff --git a/docs/Historian.md b/docs/Historian.md index ed4eedb4..9961693b 100644 --- a/docs/Historian.md +++ b/docs/Historian.md @@ -787,7 +787,7 @@ two things: | `ServerHistorian:Backend` | Advertised | Declared by | |---|---|---| -| `Gateway` | 4 — TimeAverage, Minimum, Maximum, Total | `AggregateModeMapper.Served` (derived from its mapping switch) | +| `Gateway` | 4 — TimeAverage, Minimum, Maximum, Total | `GatewayHistorianDataSource.ServedAggregates` (today `AggregateModeMapper.Served`, derived from its mapping switch) | | `SqlServer`, `QuestDb` | 17 — the [aggregate table](#aggregates-processed-reads) | `AggregateFamilies.Served` (Sql.Common; every member with a family) | | historian disabled (Null source) | **none** | `NullHistorianDataSource` | 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 c8ac094a..95b8067d 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 @@ -132,12 +132,16 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis /// public bool SupportsReverseEvents => false; + /// + /// The aggregates the Gateway backend serves — the single static the advertised AggregateFunctions folders + /// and the server's up-front BadAggregateNotSupported read through (#682). + /// Today: , the members the gateway has a retrieval mode for. + /// + public static readonly IReadOnlySet ServedAggregates = AggregateModeMapper.Served; + /// - /// - /// — the aggregates the gateway has a retrieval mode for (#682); every - /// other member answers BadAggregateNotSupported and is not advertised. - /// - public IReadOnlySet SupportedAggregates => AggregateModeMapper.Served; + /// ; every other member answers BadAggregateNotSupported and is not advertised. + public IReadOnlySet SupportedAggregates => ServedAggregates; /// /// diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/Mapping/AggregateModeMapperTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/Mapping/AggregateModeMapperTests.cs index a468a297..bfb7a527 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/Mapping/AggregateModeMapperTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/Mapping/AggregateModeMapperTests.cs @@ -50,7 +50,8 @@ public sealed class AggregateModeMapperTests var source = new GatewayHistorianDataSource( new FakeHistorianGatewayClient(), Microsoft.Extensions.Logging.Abstractions.NullLogger.Instance); - Assert.Same(AggregateModeMapper.Served, source.SupportedAggregates); + Assert.Same(GatewayHistorianDataSource.ServedAggregates, source.SupportedAggregates); + Assert.Same(AggregateModeMapper.Served, GatewayHistorianDataSource.ServedAggregates); } [Fact] From c019d0fc2b47938dd7f32c46bdc1cbd77d6b3b48 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Mon, 12 Oct 2026 09:18:53 -0400 Subject: [PATCH 3/3] fix(historian): trim only HistoryServerCapabilities/AggregateFunctions; null set = all (#682 review) ServerCapabilities/AggregateFunctions (i=2997) is Part 5's list of every aggregate the server supports, and live AggregateFilter subscriptions still compute all 37 SDK aggregates, so it keeps the SDK's full set; only i=11201 follows the backend. A null SupportedAggregates is treated as the interface default (all) by both the advertisement and HistoryRead-Processed (it used to NRE inside the batch). Docs: i=2997/i=11201 split and the boot window. Co-Authored-By: Claude --- CLAUDE.md | 2 +- docs/Historian.md | 49 +++++++----- .../Historian/IHistorianDataSource.cs | 10 ++- .../AdvertisedAggregateFunctions.cs | 76 +++++++++---------- .../OtOpcUaNodeManager.cs | 23 +++--- .../OtOpcUaSdkServer.cs | 7 +- .../AdvertisedAggregateFunctionsWireTests.cs | 45 +++++++---- .../NodeManagerHistoryReadTests.cs | 46 ++++++++++- 8 files changed, 165 insertions(+), 93 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index 6c1f9d0d..2cd77cb4 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -1353,7 +1353,7 @@ historian tagname, which is the tag's driver `FullName`. - **SQL Server** (`Driver.Historian.SqlServer`): `hist.Sample` PK `(TagId, SourceTimeUtc)` — writes are insert-if-absent, so outbox redelivery is idempotent. Stores every value kind with its **real status**; per-sample aggregates (10, simple definitions) are pushed down to T-SQL with bucket math in **100 ns ticks**; the Part 13 time-weighted seven (TimeAverage/TimeAverage2/Total/Total2/Interpolative, #671 — ⚠️ `Total` is now the time-weighted integral, no longer `SUM` — plus the **duration-based** PercentGood/PercentBad, #680) stream window + 4 bound rows into the shared `TimeWeightedAggregator` (both SQL backends, identical results; reverse `Interpolative` anchored at the later time; window capped by `ServerHistorian:MaxTimeWeightedRawRows` ⇒ `BadTooManyOperations`, percentages included). ⚠️ **#680 status rules (client-visible):** TimeAverage/Total are Uncertain if any part of the interval uses an Uncertain value, is bridged across a Bad sample, is extrapolated past the last sample (⇒ a live tag's latest interval is Uncertain) or is not covered; TimeAverage2/Total2 keep the SDK time-based rule (Uncertain time counts as Good) but never extrapolate past the last sample — that interval is Partial, later ones `BadNoData` (a report-by-exception tag with nothing after its last sample reads `BadNoData` until its next sample arrives). `docs/Historian.md` §"Time-weighted aggregates". ⚠️ **#680 per-sample (client-visible):** `Count` counts **Good** samples only (value-based status; an all-Bad interval is generic Bad, value null) and an empty interval **inside the tag's data range** is `0`/Good (outside it still `BadNoData`); `Start`/`End` return the **raw** first/last sample of any quality — its own timestamp and status, value null when Bad; `Delta` is Uncertain only when the interval's first or last sample is unusable; and every per-sample aggregate but `Average` sets **end-of-data Partial** on the interval holding the last sample (forward incl. its earlier edge, reverse strictly inside), never on a Bad result. The data-range lookup is a seek on SQL Server but a partition scan on QuestDB (unindexed `tag` SYMBOL). `docs/Historian.md` §"Aggregates". Docker-dev recipe, schema and aggregate semantics: `docs/Historian.md` §"SQL Server backend". Integration tests need `HISTORIAN_SQL_TEST_CONNECTION`. **Retention (#667):** opt-in `SqlServer:RetentionDays` (+ `PurgeInterval` 1h, `PurgeBatchSize` 4000) -> `HistorianRetentionService` purges old `hist.Sample` (per tag) / `hist.Event` in `DELETE TOP(n)` batches, Primary-gated on `IHistorizationRoleView`, one purger per DB via `sp_getapplock`; counter `otopcua.historian.retention.purged{table}`. Optional `IHistorianBackendProvider.CreateRetention` (default null). `docs/Historian.md` §Retention. - **QuestDB** (`Driver.Historian.QuestDb`, `ServerHistorian:QuestDb:*`, Npgsql over PG wire; docker-dev overlay `docker-compose.questdb.yml`): same kinds/status/aggregates/HistoryUpdate as SqlServer, soft delete by tombstone (INSERT only — never `UPDATE`/`DELETE`; every read filters `NOT deleted`). ⚠️ **Timestamps are µs** (`QuestDbConnection.Round`/`BindTimestamp`; `IHistorianDataSource.TimestampResolutionTicks` = 10 steps HistoryRead paging), **pre-1970 cannot be stored** (writer drops, HistoryUpdate `BadOutOfRange`), every write waits for WAL visibility, TTL is enforced on write commits only, set Npgsql `Maximum Pool Size` < QuestDB's 64. `docs/Historian.md` §"QuestDB backend". - ⚠️ **Reverse processed reads are Part 13/SDK-anchored (#677, client-visible):** every aggregate on SqlServer/QuestDB now anchors a reverse read at its later time — window `(end, start]`, partial interval earliest, each stamped with its later edge (was: earlier-anchored, stamped earlier edge); on forward reads Partial also marks the interval the data starts inside (not Average — the SDK's rule, forward-only); a non-default client `AggregateConfiguration` ⇒ `BadAggregateConfigurationRejected`; Gateway provisioning sends `IntegralDivisor = 1`. SDK-parity tests: `SdkAggregateParityTests`. `docs/Historian.md` §Aggregates. -- **Advertised aggregates (#682):** `IHistorianDataSource.SupportedAggregates` (Gateway 4 via `GatewayHistorianDataSource.ServedAggregates`, SqlServer/QuestDB 17 via `AggregateFamilies.Served`, Null **none**) is the one authority for both `AggregateFunctions` folders (i=2997, i=11201 — only the folder references are trimmed; the SDK's calculator factories stay for live `AggregateFilter` subscriptions) and the up-front `BadAggregateNotSupported` in HistoryRead-Processed. ⚠️ Historian disabled ⇒ Processed is now `BadAggregateNotSupported`, not `GoodNoData`. `docs/Historian.md` §Advertised aggregates. +- **Advertised aggregates (#682):** `IHistorianDataSource.SupportedAggregates` (Gateway 4 via `GatewayHistorianDataSource.ServedAggregates`, SqlServer/QuestDB 17 via `AggregateFamilies.Served`, Null **none**; a null set ⇒ all) is the one authority for `HistoryServerCapabilities/AggregateFunctions` (i=11201 — only its folder references are trimmed) and the up-front `BadAggregateNotSupported` in HistoryRead-Processed. ⚠️ `ServerCapabilities/AggregateFunctions` (i=2997) is deliberately NOT trimmed: it keeps the SDK's 37 because live `AggregateFilter` subscriptions still compute them all. ⚠️ Historian disabled ⇒ Processed is now `BadAggregateNotSupported`, not `GoodNoData`. `docs/Historian.md` §Advertised aggregates. - **Real status fix:** the recorder used to stamp every sample Good (`192`); it now carries the driver's status. ⚠️ On the **Gateway** backend this changes nothing stored: the gateway discards quality (C-002), so AVEVA still stores Good. - **Comms loss is historized as an outage marker** (`Empty` + `BadNoCommunication`, one per tag per outage) on SqlServer; dropped (`DroppedUnsupportedKind`) on Gateway. Outbox records are **format v2** — drain or delete `ContinuousHistorization:OutboxPath` before downgrading a node. - **HistoryUpdate** (SqlServer + QuestDb; Gateway ⇒ `BadHistoryOperationUnsupported`): Insert/Replace/Update/DeleteRaw/DeleteAtTime, **Primary-only** since #674 (off-Primary ⇒ `BadInvalidState`, after authorization AND after planning — an unservable request (Remove / `IsDeleteModified` / not historized) stays `BadHistoryOperationUnsupported` off-Primary; the verdict is `IHistoryUpdateRoleView` = the data-plane verdict but **closed until this node's first redundancy snapshot**, so a pair cannot both accept in the boot window; admin-only nodes register no view ⇒ no gate; reads ungated), items with a `SourceTimestamp` past now + `ServerHistorian:HistoryUpdateMaxFutureSkew` (default 1 day, ≤0 disables) ⇒ `BadOutOfRange` in the node manager (every backend), gated in both ACL modes — flag off needs the `HistoryUpdate` role from a **Cluster-scoped** grant (+ key re-check), flag on uses `DataPlaneAuthorizer`'s HistoryUpdate operation; `WriteOperate` never implies it. The `HistoryWrite` AccessLevel bit is set only when an updater is wired at materialise time. `docs/Historian.md` §HistoryUpdate, `docs/security.md`. ⚠️ **SqlServer schema v1 is accepted (#674)** with `AutoCreateSchema=false`: writes/reads/events/retention work; modified reads + HistoryUpdate ⇒ node-level `BadHistoryOperationUnsupported` on every arm (the updater throws `NotSupportedException`, nothing sent; the `HistoryWrite` bit is still advertised — not knowable at materialise time), retention skips `SampleModified`, one Warning per boot (`SqlServerSchemaBootstrapper.SupportsModifiedTrail`; `RequiredVersion` = 1, `ModifiedTrailVersion` = 2). QuestDB: keep `cairo.ttl.use.wall.clock=true`; a QuestDB trail INSERT retried after a client timeout duplicates (no DEDUP key). diff --git a/docs/Historian.md b/docs/Historian.md index 9961693b..6b8af02c 100644 --- a/docs/Historian.md +++ b/docs/Historian.md @@ -96,7 +96,7 @@ HistoryRead is served from any node (the database is shared). HistoryUpdate is * The `ServerHistorian` section in `appsettings.json` controls the historian read path. `Enabled` defaults to `false`; when disabled, the server registers `NullHistorianDataSource` and Raw / AtTime / Events HistoryRead calls on historized nodes return `GoodNoData` (empty, not an error); a -Processed read is `BadAggregateNotSupported` and no aggregate functions are advertised +Processed read is `BadAggregateNotSupported` and no history aggregate functions are advertised ([#682](#advertised-aggregates-aggregatefunctions-682)). ```json @@ -778,34 +778,45 @@ NumberOfTransitions, AnnotationCount, StandardDeviationPopulation, VariancePopul Each backend declares the aggregates it serves as `IHistorianDataSource.SupportedAggregates` — **the single authority** for two things: -- **What the server advertises.** `Server/ServerCapabilities/AggregateFunctions` (`i=2997`) and - `HistoryServerCapabilities/AggregateFunctions` (`i=11201`) — two separate folder nodes that organize the same - aggregate-function objects — list exactly that set. Before #682 both listed the SDK's full 37 regardless of backend. +- **What the server advertises as HISTORY aggregates.** `HistoryServerCapabilities/AggregateFunctions` (`i=11201`) lists + exactly that set. Before #682 it listed the SDK's full 37 regardless of backend. - **What HistoryRead-Processed refuses up front.** An aggregate outside the set is `BadAggregateNotSupported` for that node without a backend call (the backend still throws `HistoryAggregateNotSupportedException` for one as defence in depth — - derived from the same table, never a second list). + derived from the same table, never a second list). A source whose `SupportedAggregates` is `null` (out-of-tree, against + the contract) is treated as the interface default — **all** — by both consumers. -| `ServerHistorian:Backend` | Advertised | Declared by | -|---|---|---| -| `Gateway` | 4 — TimeAverage, Minimum, Maximum, Total | `GatewayHistorianDataSource.ServedAggregates` (today `AggregateModeMapper.Served`, derived from its mapping switch) | -| `SqlServer`, `QuestDb` | 17 — the [aggregate table](#aggregates-processed-reads) | `AggregateFamilies.Served` (Sql.Common; every member with a family) | -| historian disabled (Null source) | **none** | `NullHistorianDataSource` | +⚠️ **`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 +still computes all of them for live data: a MonitoredItem `AggregateFilter` is validated and computed by the SDK's +`AggregateManager`, not the historian. The SDK models the same split (`DiagnosticsNodeManager.AddAggregateFunction` always adds +to i=2997 and to i=11201 only for historical aggregates). The two are separate folder nodes organizing the same +aggregate-function objects. A client choosing a history aggregate should read **i=11201**. + +| `ServerHistorian:Backend` | i=11201 (history) | i=2997 (server) | Declared by | +|---|---|---|---| +| `Gateway` | 4 — TimeAverage, Minimum, Maximum, Total | all 37 | `GatewayHistorianDataSource.ServedAggregates` (today `AggregateModeMapper.Served`, derived from its mapping switch) | +| `SqlServer`, `QuestDb` | 17 — the [aggregate table](#aggregates-processed-reads) | all 37 | `AggregateFamilies.Served` (Sql.Common; every member with a family) | +| historian disabled (Null source) | **none** | all 37 | `NullHistorianDataSource` | ⚠️ **Client-visible: with no historian configured a processed HistoryRead is now `BadAggregateNotSupported`, not -`GoodNoData`.** An aggregate is a statistic the server cannot compute without a history, and the empty folders now say so; +`GoodNoData`.** An aggregate is a statistic the server cannot compute without a history, and the empty i=11201 now says so; Raw / AtTime / Events reads on the Null source still return the empty `GoodNoData` result. How it is wired: `OtOpcUaSdkServer` keeps the SDK's `CreateAggregateManager` (all 37 calculator factories) and then trims -only the folder ⇄ aggregate `Organizes` references to the node manager's current source — the Null default at boot; the -Host's `SetHistorianDataSource` (the backend resolved once from `ServerHistorian:Backend`) re-syncs it right after start, -and every later assignment re-syncs both directions. ⚠️ **The calculator factories are deliberately NOT removed**: a -live-data MonitoredItem with an `AggregateFilter` is validated and computed by the SDK's `AggregateManager`, not the -historian, so live aggregate subscriptions keep working for every SDK aggregate, advertised or not (pinned over the wire -by `AdvertisedAggregateFunctionsWireTests`). The aggregate-function object nodes themselves stay readable. A boot log line -names the advertised set. +only the i=11201 ⇄ aggregate `Organizes` references (both directions) to the node manager's current source — the Null +default at boot; the Host's `SetHistorianDataSource` (the backend resolved once from `ServerHistorian:Backend`) re-syncs it +right after start, and every later assignment re-syncs both ways. ⚠️ **The calculator factories are deliberately NOT +removed**: live aggregate subscriptions keep working for every SDK aggregate (pinned over the wire by +`AdvertisedAggregateFunctionsWireTests`). The aggregate-function object nodes themselves stay readable. A boot log line +names the advertised history set. + +⚠️ **Boot window.** `SetHistorianDataSource` runs right after the SDK server has started (`OtOpcUaServerHostedService.StartAsync`), +so for the few milliseconds between the endpoint opening and that call i=11201 is empty and a processed HistoryRead is +refused `BadAggregateNotSupported` — the Null default's answer. A client that browses i=11201 in that window sees an empty +list; re-browse after connect if it matters. Not changed: the `HistoryServerCapabilities` booleans (`AccessHistoryDataCapability`, `InsertDataCapability`, …) are still -whatever the SDK sets — #682 touches only the two folders (follow-up candidate). +whatever the SDK sets — #682 touches only the i=11201 folder (follow-up candidate). ### Known limitations diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/Historian/IHistorianDataSource.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/Historian/IHistorianDataSource.cs index 803a4012..2cba63ed 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/Historian/IHistorianDataSource.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/Historian/IHistorianDataSource.cs @@ -176,9 +176,10 @@ public interface IHistorianDataSource : IDisposable /// /// The Part 13 aggregates this backend serves for a processed HistoryRead (Gitea #682) — the ONE authority for - /// both what the server advertises under ServerCapabilities/AggregateFunctions (i=2997) and - /// HistoryServerCapabilities/AggregateFunctions (i=11201), and what its HistoryRead-Processed answers - /// BadAggregateNotSupported up front without calling the backend. + /// both what the server advertises under HistoryServerCapabilities/AggregateFunctions (i=11201) and what + /// its HistoryRead-Processed answers BadAggregateNotSupported up front without calling the backend. + /// (ServerCapabilities/AggregateFunctions, i=2997, is not affected: it lists every aggregate the server + /// computes, including for live AggregateFilter subscriptions.) /// /// /// The default is — "every aggregate the server can name; a member I cannot @@ -187,7 +188,8 @@ public interface IHistorianDataSource : IDisposable /// null "all" sentinel so every consumer handles one shape. Every PRODUCTION backend declares its set itself /// (a reflection guard pins that): a wrapper that forgot to forward this would otherwise advertise every aggregate. /// The set is read once when the source is wired (the backend is fixed at boot by ServerHistorian:Backend) - /// and again on every processed read, so it must be cheap and stable — return a static frozen set. + /// and again on every processed read, so it must be cheap and stable — return a static frozen set. A source that + /// returns null anyway is treated as the default (all) by both consumers. /// IReadOnlySet SupportedAggregates => HistoryAggregateTypes.All; diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/AdvertisedAggregateFunctions.cs b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/AdvertisedAggregateFunctions.cs index dd475de4..59b256a6 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/AdvertisedAggregateFunctions.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/AdvertisedAggregateFunctions.cs @@ -6,24 +6,23 @@ using ZB.MOM.WW.OtOpcUa.Core.Abstractions; namespace ZB.MOM.WW.OtOpcUa.OpcUaServer; /// -/// Gitea #682 — make the two advertised aggregate-function folders, ServerCapabilities/AggregateFunctions -/// (i=2997) and HistoryServerCapabilities/AggregateFunctions (i=11201), list exactly the aggregates the wired -/// historian backend serves (), instead of the SDK's full -/// registered set. +/// Gitea #682 — make HistoryServerCapabilities/AggregateFunctions (i=11201) list exactly the aggregates the +/// wired historian backend serves for HistoryRead (), instead of +/// the SDK's full registered set. /// /// /// -/// The SDK's AggregateManager.RegisterFactory both registers a calculator factory AND adds an -/// Organizes reference from each folder to the aggregate's object node. Only the REFERENCES are trimmed -/// here: the factories stay, because a live-data MonitoredItem with an AggregateFilter is validated -/// against (and computed by) the AggregateManager, not the historian — removing a factory would break live -/// aggregate subscriptions for every aggregate the historian happens not to serve. +/// ⚠️ ServerCapabilities/AggregateFunctions (i=2997) is deliberately NOT trimmed. Part 5 defines it as every +/// aggregate the SERVER supports, and the server still supports all the SDK's registered aggregates for live +/// data: a MonitoredItem AggregateFilter is validated and computed by the SDK's AggregateManager, +/// not the historian. The SDK itself models the split — DiagnosticsNodeManager.AddAggregateFunction always +/// adds to i=2997 and adds to i=11201 only when the aggregate is historical. i=11201 is the history surface, so it +/// is the one that follows the backend. /// /// -/// The two folders are separate nodes that reference the same aggregate object nodes (both directions). The -/// aggregate nodes themselves are kept (they are standard type-system nodes a client may still Read); only the -/// folder ⇄ aggregate Organizes pair is removed or restored. Idempotent and reversible, so re-wiring a -/// different source re-syncs both folders. +/// Only the folder ⇄ aggregate Organizes reference pair is removed or restored (both directions); the +/// calculator factories and the aggregate-function object nodes stay. Idempotent and reversible, so re-wiring a +/// different source re-syncs the folder. /// /// internal static class AdvertisedAggregateFunctions @@ -35,20 +34,26 @@ internal static class AdvertisedAggregateFunctions .Select(f => (NodeId)f.GetValue(null)!) .ToArray(); - /// The two advertised folders (both are trimmed — they are separate nodes). - internal static readonly NodeId[] Folders = - [ - ObjectIds.Server_ServerCapabilities_AggregateFunctions, - ObjectIds.HistoryServerCapabilities_AggregateFunctions, - ]; + /// The trimmed folder — the HISTORY aggregate list only (see remarks for why not i=2997). + internal static readonly NodeId Folder = ObjectIds.HistoryServerCapabilities_AggregateFunctions; /// - /// Sync both folders to : an aggregate-function node is organized under a folder iff the - /// server maps it () AND contains it. + /// The set a source serves, treating a null (an + /// out-of-tree source that returns null despite the non-nullable contract) as , + /// the interface default — the ONE null rule both the advertisement and HistoryRead-Processed use. + /// + /// The historian source. + /// Its served set, never null. + public static IReadOnlySet ServedBy(IHistorianDataSource source) => + source.SupportedAggregates ?? HistoryAggregateTypes.All; + + /// + /// Sync i=11201 to : an aggregate-function node is organized under it iff the server maps + /// it () AND contains it. /// /// The running server. /// The backend's served aggregates. - /// The advertised NodeIds, or null when the diagnostics node manager is not available yet. + /// The advertised NodeIds, or null when the diagnostics node manager or folder is not available. public static IReadOnlyList? Apply(IServerInternal server, IReadOnlySet served) { if (server?.DiagnosticsNodeManager is not CustomNodeManager2 diagnostics) return null; @@ -56,10 +61,8 @@ internal static class AdvertisedAggregateFunctions var advertised = new List(); lock (diagnostics.Lock) { - var folders = Folders - .Select(id => diagnostics.FindPredefinedNode(id, typeof(BaseObjectState))) - .Where(f => f is not null) - .ToList(); + var folder = diagnostics.FindPredefinedNode(Folder, typeof(BaseObjectState)); + if (folder is null) return null; foreach (var aggregateId in StandardAggregateIds) { @@ -69,19 +72,16 @@ internal static class AdvertisedAggregateFunctions var advertise = AggregateFunctionMap.Map(aggregateId) is { } aggregate && served.Contains(aggregate); if (advertise) advertised.Add(aggregateId); - foreach (var folder in folders) + var linked = folder.ReferenceExists(ReferenceTypeIds.Organizes, false, aggregateId); + if (advertise && !linked) { - var linked = folder!.ReferenceExists(ReferenceTypeIds.Organizes, false, aggregateId); - if (advertise && !linked) - { - folder.AddReference(ReferenceTypeIds.Organizes, false, aggregateId); - aggregateNode.AddReference(ReferenceTypeIds.Organizes, true, folder.NodeId); - } - else if (!advertise && linked) - { - folder.RemoveReference(ReferenceTypeIds.Organizes, false, aggregateId); - aggregateNode.RemoveReference(ReferenceTypeIds.Organizes, true, folder.NodeId); - } + folder.AddReference(ReferenceTypeIds.Organizes, false, aggregateId); + aggregateNode.AddReference(ReferenceTypeIds.Organizes, true, folder.NodeId); + } + else if (!advertise && linked) + { + folder.RemoveReference(ReferenceTypeIds.Organizes, false, aggregateId); + aggregateNode.RemoveReference(ReferenceTypeIds.Organizes, true, folder.NodeId); } } } diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs index eba945fc..3ae1fe4e 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs @@ -345,7 +345,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 /// node-manager Lock, so the override may block-bridge to this (async) source. /// /// - /// Gitea #682: assigning a source also re-syncs the advertised AggregateFunctions folders to its + /// Gitea #682: assigning a source also re-syncs HistoryServerCapabilities/AggregateFunctions to its /// (see ). /// /// @@ -360,9 +360,10 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 } /// - /// Gitea #682 — trim ServerCapabilities/AggregateFunctions (i=2997) and - /// HistoryServerCapabilities/AggregateFunctions (i=11201) to the current source's - /// . Called on every + /// Gitea #682 — trim HistoryServerCapabilities/AggregateFunctions (i=11201) to the current source's + /// (null ⇒ all, ). + /// ServerCapabilities/AggregateFunctions (i=2997) keeps the SDK's full set — the server still computes every + /// one of them for live AggregateFilter subscriptions. Called on every /// assignment and by right after the SDK registers its aggregate factories, so the /// Null default is applied from boot. Only the folder references change — the AggregateManager's calculator /// factories (live AggregateFilter subscriptions) are untouched. Never throws: a failure leaves the previous @@ -373,11 +374,11 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 try { var source = _historianDataSource; - var advertised = AdvertisedAggregateFunctions.Apply(Server, source.SupportedAggregates); + var advertised = AdvertisedAggregateFunctions.Apply(Server, AdvertisedAggregateFunctions.ServedBy(source)); if (advertised is not null) { _logger.LogInformation( - "OtOpcUaNodeManager: advertising {Count} aggregate function(s) for historian {Source}: {Aggregates}", + "OtOpcUaNodeManager: advertising {Count} history aggregate function(s) for historian {Source}: {Aggregates}", advertised.Count, source.GetType().Name, string.Join(", ", advertised.Select(id => AggregateFunctionMap.Map(id)?.ToString() ?? id.ToString()))); } @@ -5091,7 +5092,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 // #677: a client AggregateConfiguration other than the server defaults is refused per node, never silently ignored. var configurationServable = ServerAggregateConfiguration.IsServable(details.AggregateConfiguration); nodesToProcess = AuthorizeHistoryRead(context, nodesToProcess, results, errors); - var source = HistorianDataSource; + var served = AdvertisedAggregateFunctions.ServedBy(HistorianDataSource); WithHistoryReadBatchLimiter(nodesToProcess, results, errors, () => { using var deadline = CreateDeadline(); @@ -5102,11 +5103,11 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 // AggregateType is a per-node parallel collection (same length as nodesToRead, enforced by // the base dispatcher). handle.Index is the node's position in that collection. var aggregateNodeId = details.AggregateType[handle.Index]; - // #682: one authority — the source's SupportedAggregates (also what the AggregateFunctions folders - // advertise). An aggregate the server cannot name, or the backend does not serve, is refused up front - // without a backend call. + // #682: one authority — the source's SupportedAggregates (also what HistoryServerCapabilities/ + // AggregateFunctions advertises; null ⇒ all, the same rule). An aggregate the server cannot name, or the + // backend does not serve, is refused up front without a backend call. var aggregate = AggregateFunctionMap.Map(aggregateNodeId); - if (aggregate is null || !source.SupportedAggregates.Contains(aggregate.Value)) + if (aggregate is null || !served.Contains(aggregate.Value)) { errors[handle.Index] = StatusCodes.BadAggregateNotSupported; continue; diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaSdkServer.cs b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaSdkServer.cs index 67af4072..eb5a88da 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaSdkServer.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaSdkServer.cs @@ -202,9 +202,10 @@ public sealed class OtOpcUaSdkServer : StandardServer /// /// Gitea #682 — keep the SDK's full set of calculator factories (live-data AggregateFilter subscriptions - /// need them), then trim the advertised AggregateFunctions folders to the node manager's current historian - /// source. At this point that is the Null default (no aggregates); the Host's - /// re-syncs to the configured backend's set right after start. + /// need them, and ServerCapabilities/AggregateFunctions keeps listing all of them), then trim + /// HistoryServerCapabilities/AggregateFunctions to the node manager's current historian source. At this + /// point that is the Null default (no history aggregates); the Host's + /// re-syncs to the configured backend's set right after start. /// /// protected override AggregateManager CreateAggregateManager( diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/AdvertisedAggregateFunctionsWireTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/AdvertisedAggregateFunctionsWireTests.cs index 8c7cb4f6..e3596cc1 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/AdvertisedAggregateFunctionsWireTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/AdvertisedAggregateFunctionsWireTests.cs @@ -12,11 +12,11 @@ using static ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests.ServerEventRig; namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests; /// -/// Gitea #682 over a real SDK client session: ServerCapabilities/AggregateFunctions (i=2997) and -/// HistoryServerCapabilities/AggregateFunctions (i=11201) list exactly the aggregates the wired historian -/// backend serves — the data sources are built through exactly as the Host -/// builds them — while a live-data AggregateFilter subscription keeps working for every aggregate the SDK -/// calculates, advertised or not. +/// Gitea #682 over a real SDK client session: HistoryServerCapabilities/AggregateFunctions (i=11201) lists +/// exactly the aggregates the wired historian backend serves — the data sources are built through +/// exactly as the Host builds them — while +/// ServerCapabilities/AggregateFunctions (i=2997) keeps the SDK's full registered set in every configuration, +/// because a live-data AggregateFilter subscription keeps working for every one of them. /// public sealed class AdvertisedAggregateFunctionsWireTests { @@ -28,6 +28,17 @@ public sealed class AdvertisedAggregateFunctionsWireTests "PercentBad", "PercentGood", "Range", "Start", "TimeAverage", "TimeAverage2", "Total", "Total2", ]; + /// The SDK's 37 registered aggregates (StandardServer.CreateAggregateManager) — i=2997, always. + private static readonly string[] AllSdkNames = + [ + "AnnotationCount", "Average", "Count", "Delta", "DeltaBounds", "DurationBad", "DurationGood", "DurationInStateNonZero", + "DurationInStateZero", "End", "EndBound", "Interpolative", "Maximum", "Maximum2", "MaximumActualTime", + "MaximumActualTime2", "Minimum", "Minimum2", "MinimumActualTime", "MinimumActualTime2", "NumberOfTransitions", + "PercentBad", "PercentGood", "Range", "Range2", "StandardDeviationPopulation", "StandardDeviationSample", "Start", + "StartBound", "TimeAverage", "TimeAverage2", "Total", "Total2", "VariancePopulation", "VarianceSample", + "WorstQuality", "WorstQuality2", + ]; + private static IHistorianDataSource Backend(string backend) { var config = new ConfigurationBuilder().AddInMemoryCollection(new Dictionary @@ -47,7 +58,7 @@ public sealed class AdvertisedAggregateFunctionsWireTests [InlineData("Gateway")] [InlineData("SqlServer")] [InlineData("QuestDb")] - public async Task Both_folders_advertise_exactly_the_backend_set(string backend) + public async Task History_folder_advertises_the_backend_set_and_the_server_folder_keeps_every_aggregate(string backend) { var ct = TestContext.Current.CancellationToken; var rig = await StartAsync($"AggAdv{backend}", encrypted: false, ct); @@ -58,14 +69,14 @@ public sealed class AdvertisedAggregateFunctionsWireTests 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); + (await AdvertisedAsync(session, ObjectIds.Server_ServerCapabilities_AggregateFunctions, ct)).ShouldBe(AllSdkNames); } finally { await rig.DisposeAsync(); } } [Fact] - public async Task With_no_historian_wired_both_folders_are_empty_from_boot() + public async Task With_no_historian_wired_the_history_folder_is_empty_from_boot() { // The Null default is applied by the CreateAggregateManager override, before any source is assigned. var ct = TestContext.Current.CancellationToken; @@ -75,8 +86,8 @@ public sealed class AdvertisedAggregateFunctionsWireTests 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(); + (await AdvertisedAsync(session, ObjectIds.Server_ServerCapabilities_AggregateFunctions, ct)).ShouldBe(AllSdkNames); } finally { await rig.DisposeAsync(); } } @@ -92,16 +103,18 @@ public sealed class AdvertisedAggregateFunctionsWireTests 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 AdvertisedAsync(session, ObjectIds.HistoryServerCapabilities_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. + // Gateway has no arithmetic Average: dropped from the history folder (inverse reference too), kept in the + // server folder. 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(); + (await AdvertisedAsync(session, ObjectIds.Server_ServerCapabilities_AggregateFunctions, ct)).ShouldBe(AllSdkNames); + (await ParentsAsync(session, ObjectIds.AggregateFunction_Average, ct)).ShouldBe( + [ObjectIds.Server_ServerCapabilities_AggregateFunctions]); // And back: removed references are restored. rig.Nm.HistorianDataSource = Backend("QuestDb"); @@ -111,7 +124,7 @@ public sealed class AdvertisedAggregateFunctionsWireTests } [Theory] - [InlineData("StandardDeviationSample")] // served by no historian backend, never advertised + [InlineData("StandardDeviationSample")] // served by no historian backend, never in the history folder [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) { @@ -128,8 +141,10 @@ public sealed class AdvertisedAggregateFunctionsWireTests 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)) + (await AdvertisedAsync(session, ObjectIds.HistoryServerCapabilities_AggregateFunctions, ct)) .ShouldNotContain(aggregateName); + (await AdvertisedAsync(session, ObjectIds.Server_ServerCapabilities_AggregateFunctions, ct)) + .ShouldContain(aggregateName); // i=2997 tells the truth: the live aggregate IS supported var subscription = new Subscription(session.DefaultSubscription) { PublishingInterval = 100 }; session.AddSubscription(subscription); 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 d119e496..e8a231ed 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadTests.cs @@ -326,6 +326,48 @@ public sealed class NodeManagerHistoryReadTests : IDisposable await host.DisposeAsync(); } + /// + /// #682 review: an out-of-tree source returning a NULL (despite + /// the non-nullable contract) is treated as the interface default — ALL — by BOTH consumers: the processed read + /// dispatches (no NullReferenceException failing the whole request) and HistoryServerCapabilities/AggregateFunctions + /// advertises every mapped aggregate. + /// + [Fact] + public async Task A_null_supported_set_means_all_for_both_the_read_and_the_advertisement() + { + var (host, server) = await BootAsync(); + var nm = server.NodeManager!; + var fake = new RecordingHistorianDataSource { ProcessedResult = Empty(), Supported = null }; + nm.HistorianDataSource = fake; + + nm.EnsureVariable("eq-1/nul", parentFolderNodeId: null, displayName: "Nul", dataType: "Float", + writable: false, historianTagname: "WW.Nul", realm: AddressSpaceRealm.Uns); + var nodeId = nm.TryGetVariable("eq-1/nul")!.NodeId; + + var (_, errors) = InvokeHistoryRead(server, nm, new ReadProcessedDetails + { + StartTime = DateTime.UtcNow.AddHours(-1), + EndTime = DateTime.UtcNow, + ProcessingInterval = 10_000.0, + AggregateType = new NodeIdCollection { ObjectIds.AggregateFunction_StandardDeviationPopulation }, + }, nodeId); + StatusCode.IsBad(errors[0].StatusCode).ShouldBeFalse($"got {errors[0].StatusCode}"); + fake.LastCall.ShouldBe("Processed"); + fake.LastAggregate.ShouldBe(HistoryAggregateType.StandardDeviationPopulation); + + var diagnostics = (CustomNodeManager2)nm.Server.DiagnosticsNodeManager; + var folder = diagnostics.FindPredefinedNode(ObjectIds.HistoryServerCapabilities_AggregateFunctions, null)!; + AggregateFunctionMap.ByNodeId.Keys.Count(id => folder.ReferenceExists(ReferenceTypeIds.Organizes, false, id)) + .ShouldBe(AggregateFunctionMap.ByNodeId.Count); + + // Positive control on the same folder: a real (non-null) set trims it. + nm.HistorianDataSource = NullHistorianDataSource.Instance; + AggregateFunctionMap.ByNodeId.Keys.Count(id => folder.ReferenceExists(ReferenceTypeIds.Organizes, false, id)) + .ShouldBe(0); + + await host.DisposeAsync(); + } + /// /// #677: a client AggregateConfiguration that differs from the server defaults is refused per node with /// BadAggregateConfigurationRejected (it used to be silently ignored); the source is never invoked. @@ -674,8 +716,8 @@ public sealed class NodeManagerHistoryReadTests : IDisposable public bool ThrowOnRead { get; init; } public bool AggregateNotSupported { get; init; } public bool LimitExceeded { get; init; } - public IReadOnlySet Supported { get; init; } = HistoryAggregateTypes.All; - public IReadOnlySet SupportedAggregates => Supported; + public IReadOnlySet? Supported { get; init; } = HistoryAggregateTypes.All; + public IReadOnlySet SupportedAggregates => Supported!; public HistorianRead RawResult { get; set; } = new(Array.Empty(), null); public HistorianRead ProcessedResult { get; set; } = new(Array.Empty(), null); public HistorianRead AtTimeResult { get; set; } = new(Array.Empty(), null);