diff --git a/docs/Historian.md b/docs/Historian.md index e83742dbc..78df197e9 100644 --- a/docs/Historian.md +++ b/docs/Historian.md @@ -639,14 +639,71 @@ attributes as `DevWinPlatform.CPULoadAvg`. Both are returned by the one request served: a notifier whose source is `X` serves the events whose `SourceName` is `X` or starts with `X.`. The match is on the `.` boundary, so `DevWinPlatform` never serves `DevWinPlatform2` or `DevWinPlatform2.CPULoadAvg`. This matches live delivery, where an attribute alarm's condition fires on the same folder as the object's own events. Each event -keeps its own `SourceName` (the attribute's); its `SourceNode` is the condition its stored alarm id names, so an AVEVA -row (no OtOpcUa alarm id) has a null `SourceNode`. The attribute events are part of the notifier's own read, so they +keeps its own `SourceName` (the attribute's); its `SourceNode` is the condition its stored alarm id names, or, since #886, +the native condition that fires on the folder and dials the row's `source_conditionvariable` (below). The attribute events are part of the notifier's own read, so they are governed by that notifier's HistoryRead verdict: with `Security:Acl:Enabled`, a session denied the folder gets `BadUserAccessDenied` and the backend is never called. No attribute-level notifier is exposed; a notifier whose source is itself an attribute name (`DevWinPlatform.CPULoadAvg`) still answers `GoodNoData` from the gateway, whose `Source_Object` is the object. The SqlServer and QuestDB backends match the source exactly, as before: they hold only -OtOpcUa's own alarm events. ⚠️ A production folder's source is its NodeId identifier (a RawPath or an EquipmentId), -not a Galaxy object name, so AVEVA's own events reach a folder only where those coincide (follow-up [#886](https://gitea.dohertylan.com/dohertj2/lmxopcua/issues/886)). +OtOpcUa's own alarm events. A production folder's own source is its NodeId identifier (a RawPath or an EquipmentId), +not a Galaxy object name; #886 (below) adds the Galaxy objects behind the folder as further sources. + +#### Galaxy native event history on real folders (Gitea #886) + +**Gap.** Before #886 a notifier folder read exactly one event source, its NodeId identifier. AVEVA stores a Galaxy +object's events under the object's `tag_name` (`Source_Object`), which is never a RawPath or an EquipmentId, so AVEVA's +own alarm and event history reached no production folder. AVEVA's lower-case property keys (`priority`, `alarm_id`, +`source_conditionvariable`, and the keys of OtOpcUa's own events, which AVEVA stores lower-cased: `alarmname`, +`alarmid`, `severity`, `message`) were also missed, so every row read Severity 1 with no ConditionName. + +**Rule: a folder's history sources are the folder itself plus the Galaxy objects of the native conditions that fire on it +live.** A native Galaxy condition fires live on its Raw parent folder and on every UNS equipment folder wired to it +(`WireAlarmNotifiers`). History uses the same membership: + +- The condition's Galaxy reference is its raw tag's `attributeRef` (`tag_name.AttributeName`, the address the driver + dials, e.g. `Oven_007.TempHiAlarm`). The composer reads it with the same parser the Galaxy driver uses + (`GalaxyAttributeRef`) and only for a `GalaxyMxGateway` tag that carries an alarm (`RawTagPlan.NativeAlarmReference`). + The applier hands it to the node manager after it materialises the condition (`BindNativeAlarmHistory`). +- The object is the reference up to the first `.` (`Oven_007`). An equipment folder whose tags come from two Galaxy + objects reads both. +- When a folder is read, the server collects, under the node-manager lock, the object of every native condition whose + parent is that folder or that is wired to it. Sources are the folder's own identifier (OtOpcUa's own events, as + before) plus those objects, de-duplicated and ordinal-sorted. A folder with no Galaxy conditions keeps its single + source and the exact pre-#886 code path. +- Each object is read with the #882 rule: the object's own events and `.` rows, never a prefix + sibling (`Oven_007` never serves `Oven_0070`). This serves every alarm of the object, including one whose condition + is not wired to this folder, as #882 already does for a folder named after the object. + +**Fan-out and merge (`EventSourceFanOut`).** The historiangw `ReadEvents` filter is a single property predicate, so there +is no multi-source request and the wire contract is unchanged: one gateway request per source, sequential within a +folder, each with the page's cap. The results are merged into one stream ordered by `(EventTimeUtc, source, arrival)`: +time in the read's direction, then the source's position in the folder's ordinal-sorted list (reversed for a reverse +read), then the order the source returned them. Ties are deliberately **not** re-sorted by EventId: the Gateway has no +intra-timestamp order, and re-sorting could move an unread tie in front of one already sent; a source's own order is what +keeps the boundary skip stable, as for a single source. The merge is de-duplicated by EventId and cut to the cap. When a +source was truncated at time H, its unread rows are at or after H and, at H, follow the ones it returned, so the page keeps +everything before the earliest such H and, at H, each source in merge order up to and including the first one cut there. +That prefix is never empty. With the same cap on every source the cut to the cap already stops there; the bound matters +for an uncapped or default-capped read. The merge presents itself as one source to the existing paging machinery, so +continuation points, boundary-tie skips, the walked-back reverse read (Gateway, which sorts its ties by EventId itself) and +the tie-cluster over-fetch (a forward read, reversed for a reverse read, which is why the reverse merge is the forward merge +backwards) are unchanged. A continuation point is bound to the folder's whole source set; if a redeploy changes the set +between pages, the next page answers `BadContinuationPointInvalid`. A folder with one source is read exactly as before. + +**ACL.** Nothing new is authorized per source. Every source is read inside the folder's own HistoryRead, after +`AuthorizeHistoryRead` (#826): with `Security:Acl:Enabled`, a session denied the folder gets `BadUserAccessDenied` and no +source is read (fail closed). + +**Mapping AVEVA rows (EventMapper).** Property names are matched case-insensitively. + +| Field | Source | Notes | +|---|---|---| +| `Severity` | `priority` (AVEVA native rows) through `GalaxyAlarmPriority`, the live Galaxy mapping (`< 250` → 250, `< 500` → 500, `< 750` → 750, else 1000), else `severity` clamped to 1–1000 (OtOpcUa's own rows), else 1 | AVEVA's own `severity` (1–4, or 0 for a system event) is not on the OPC UA scale and is never used for a row that has `priority`. ⚠️ AVEVA ranks priority 1 as the most urgent and 999 as the least; the live mapping treats a higher number as more severe. History follows live so the two agree (lmxopcua#887 tracks the inversion). | +| `ConditionName` | The live condition's name when the row's `source_conditionvariable` names a native condition that fires on this folder; else `alarmname` (OtOpcUa's own rows); else the attribute part of `source_conditionvariable` (`TempHiAlarm`) | Live parity: a resolved row shows the same ConditionName as the live condition. | +| `SourceNode` | The native condition whose `attributeRef` equals the row's `source_conditionvariable` (case-insensitive, as the Galaxy driver matches it), among the conditions that fire on this folder; else the condition `alarmid` names (OtOpcUa's own rows) | A condition that does not fire on this folder is never named, so history discloses no node outside the folder's live event set. Object-level system events have none. | +| `Message` | `message` (OtOpcUa's own rows), else the event type (`Alarm.Set`, `System.Stop`) | Unchanged apart from case. | +| `AlarmId` (internal) | `alarmid`, else `alarm_id` | Used only to resolve `SourceNode`. AVEVA's `alarm_id` identifies one alarm occurrence (shared by its Set and Clear), not a condition, so it is never a ConditionName. | + Event-field projection supports the standard `BaseEventType` select clauses — `EventId`, `SourceName`, `Time`, `ReceiveTime`, `Message`, diff --git a/docs/plans/evidence/2026-11-10-opcua2/aveva-history-801.md b/docs/plans/evidence/2026-11-10-opcua2/aveva-history-801.md index 9e80333a5..d5d90dd39 100644 --- a/docs/plans/evidence/2026-11-10-opcua2/aveva-history-801.md +++ b/docs/plans/evidence/2026-11-10-opcua2/aveva-history-801.md @@ -88,6 +88,36 @@ read over SMB and never printed. - Report: [`aveva-history-801/aveva-gate-sdk2-882.json.gz`](aveva-history-801/aveva-gate-sdk2-882.json.gz), sha256 of the JSON `449e689c…9b27` (label `sdk2-882-008f0b36`: the #882 working tree on top of `008f0b36`). +## #886 re-run (2026-11-12) + +The same harness with the #886 change: a folder's event history also reads the Galaxy objects behind the native conditions +that fire on it, and AVEVA's lower-case `priority` / `alarm_id` / `source_conditionvariable` keys are mapped. Same vd03 +HistorianGateway `ea9f21c`, same event window, read-only, network-only, **no WinRM**. The API key was read over SMB from +vd03's OtOpcUa environment backup, kept in a 0600 file, passed by environment and never printed; the local copy was deleted +afterwards. + +**The new case is a realistic equipment folder.** `EQ-886-ZipsenLeft` (UNS) is not named after any Galaxy object. Two native +conditions are wired to it, as the applier wires a Galaxy deployment: `Plant/Galaxy/DevWinPlatform/CPULoadHiAlarm` +(attributeRef `DevWinPlatform.CPULoadHiAlarm`) and `Plant/Galaxy/Left_001/FRR_Warning` (attributeRef `Left_001.FRR_Warning`), +each under its own raw device folder. + +| Check | Result | +|---|---| +| Folder's own source (what it read before #886) | `EQ-886-ZipsenLeft`: **0 events** in AVEVA | +| Whole read, wire | **92 events**: `DevWinPlatform` 2 + `DevWinPlatform.CPULoadAvg` 70 + 20 `Left_001` / `Left_001.*` rows (FRR_Warning alarms, user writes, deploy events). They are exactly the unfiltered read's rows named either object or `.*`, in time order across both objects. | +| Paging | 1-event pages (92 pages) and 7-event pages (14 pages) equal the whole read, row for row. | +| Reverse | Whole and 5-event pages (19 pages): newest first, ties by EventId descending, equal to the expected order. | +| SourceNode / ConditionName | **73 rows** resolve to their live condition: 70 `CPULoadAvg` alarm rows → `Plant/Galaxy/DevWinPlatform/CPULoadHiAlarm` / `CPULoadHiAlarm`, 3 `FRR_Warning` rows → `Plant/Galaxy/Left_001/FRR_Warning` / `FRR_Warning`. Every other row (system and user events) has a null SourceNode. | +| Severity | The wire equals the backend's mapping for every row. Alarm rows (`priority` 500 / 700) read **750**; priority-999 system and user events read **1000** (live Galaxy table; see lmxopcua#887 on its direction). Before: 1 for every row. | + +`diff_runs.py` against `aveva-gate-sdk2-882.json.gz`: 226 shared cases, **221 identical**. The 5 that differ are expected: +the 4 `DevWinPlatform` event cases (same 72 events, ids, times, sources and messages; only Severity changed from 1 to +750 × 70 / 1000 × 2) and `metadata.historyServerCapabilities` (#881, merged after that baseline). Seven cases are new +(`events886.*`). Harness failures: 0. + +Report: [`aveva-history-801/aveva-gate-sdk2-886.json.gz`](aveva-history-801/aveva-gate-sdk2-886.json.gz), sha256 of the +JSON `6ca314ee…c890` (label `sdk2-886-da948e5f-wt`: the #886 working tree on top of `da948e5f`). + ## Files - [`aveva-history-801/aveva-gate-sdk2-d46044f5.json.gz`](aveva-history-801/aveva-gate-sdk2-d46044f5.json.gz), sha256 of the JSON `00b5ecb6…fc27`. diff --git a/docs/plans/evidence/2026-11-10-opcua2/aveva-history-801/aveva-gate-sdk2-886.json.gz b/docs/plans/evidence/2026-11-10-opcua2/aveva-history-801/aveva-gate-sdk2-886.json.gz new file mode 100644 index 000000000..38d224e24 Binary files /dev/null and b/docs/plans/evidence/2026-11-10-opcua2/aveva-history-801/aveva-gate-sdk2-886.json.gz differ diff --git a/scripts/validation/aveva-history-801/README.md b/scripts/validation/aveva-history-801/README.md index 83ac2286a..140b2a926 100644 --- a/scripts/validation/aveva-history-801/README.md +++ b/scripts/validation/aveva-history-801/README.md @@ -8,6 +8,9 @@ are set. It is read-only: the Gateway backend has no updater, and the only Histo `HISTGW_WRITE_SANDBOX_TAG` and is refused by the server before any backend call. Record of the 2026-11-12 run: [`docs/plans/evidence/2026-11-10-opcua2/aveva-history-801.md`](../../../docs/plans/evidence/2026-11-10-opcua2/aveva-history-801.md). +The #886 cases (`events886.*`) wire Galaxy native conditions for `OTOPCUA_AVEVA_GATE_GALAXY_REFS` (default +`DevWinPlatform.CPULoadHiAlarm,Left_001.FRR_Warning`) to one equipment folder, `EQ-886-ZipsenLeft`, that is named after +none of their objects. ## Run (SDK 2, this tree) diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Commons/OpcUa/DeferredAddressSpaceSink.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Commons/OpcUa/DeferredAddressSpaceSink.cs index 4ee8e917a..21be4c4ac 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Commons/OpcUa/DeferredAddressSpaceSink.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Commons/OpcUa/DeferredAddressSpaceSink.cs @@ -59,6 +59,12 @@ public sealed class DeferredAddressSpaceSink : IOpcUaAddressSpaceSink, ISurgical public void WireAlarmNotifiers(string alarmNodeId, AddressSpaceRealm alarmRealm, IReadOnlyList notifierFolderNodeIds, AddressSpaceRealm notifierFolderRealm) => _inner.WireAlarmNotifiers(alarmNodeId, alarmRealm, notifierFolderNodeIds, notifierFolderRealm); + /// + // Forwarded for the same reason as WireAlarmNotifiers: the interface default is a no-op, so a missing forward would + // leave every driver-role host's folders without their Galaxy event-history sources (#886). + public void BindNativeAlarmHistory(string alarmNodeId, AddressSpaceRealm alarmRealm, string alarmReference) + => _inner.BindNativeAlarmHistory(alarmNodeId, alarmRealm, alarmReference); + /// public void EnsureFolder(string folderNodeId, string? parentNodeId, string displayName, AddressSpaceRealm realm) => _inner.EnsureFolder(folderNodeId, parentNodeId, displayName, realm); diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Commons/OpcUa/IOpcUaAddressSpaceSink.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Commons/OpcUa/IOpcUaAddressSpaceSink.cs index 7c69c3147..918926023 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Commons/OpcUa/IOpcUaAddressSpaceSink.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Commons/OpcUa/IOpcUaAddressSpaceSink.cs @@ -134,6 +134,21 @@ public interface IOpcUaAddressSpaceSink void WireAlarmNotifiers(string alarmNodeId, AddressSpaceRealm alarmRealm, IReadOnlyList notifierFolderNodeIds, AddressSpaceRealm notifierFolderRealm); + /// + /// Gitea #886 — record the upstream (AVEVA) reference of an already-materialised NATIVE alarm condition: the + /// Galaxy alarm reference (tag_name.AlarmAttribute) the backing raw tag dials. AVEVA stores that alarm's + /// history, and its object's own events, under the reference's object (tag_name), so every folder the + /// condition fires on live (its parent and each folder wired) also reads that + /// object's event history, and an AVEVA history row naming this reference resolves its SourceNode and + /// ConditionName to this condition. Call it after ; idempotent; a + /// condition that is not materialised is a no-op. A rebuild or the condition's removal forgets the reference. + /// The default does nothing (a sink with no event history). + /// + /// The native alarm condition's node id (== the backing tag's RawPath). + /// The namespace realm the condition lives in (Raw for native alarms). + /// The Galaxy alarm reference, e.g. Oven_007.TempHiAlarm. + void BindNativeAlarmHistory(string alarmNodeId, AddressSpaceRealm alarmRealm, string alarmReference) { } + /// /// Ensure a folder node exists under the given parent. Used by AddressSpaceApplier to /// materialise the UNS Area/Line/Equipment hierarchy in the address space. When @@ -288,6 +303,9 @@ public sealed class NullOpcUaAddressSpaceSink : IOpcUaAddressSpaceSink /// public void WireAlarmNotifiers(string alarmNodeId, AddressSpaceRealm alarmRealm, IReadOnlyList notifierFolderNodeIds, AddressSpaceRealm notifierFolderRealm) { } + /// + public void BindNativeAlarmHistory(string alarmNodeId, AddressSpaceRealm alarmRealm, string alarmReference) { } + /// public void EnsureFolder(string folderNodeId, string? parentNodeId, string displayName, AddressSpaceRealm realm) { } diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/GalaxyAlarmPriority.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/GalaxyAlarmPriority.cs new file mode 100644 index 000000000..b475e14dd --- /dev/null +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/GalaxyAlarmPriority.cs @@ -0,0 +1,37 @@ +namespace ZB.MOM.WW.OtOpcUa.Core.Abstractions; + +/// +/// The Galaxy alarm priority → OPC UA Part 9 severity mapping, shared by live alarms (the Galaxy driver's +/// MxAccessSeverityMapper) and AVEVA alarm history (priority, Gitea #886) so the two agree. +/// +/// +/// The four-bucket ladder (250 / 500 / 750 / 1000 — Low / Medium / High / Critical) is the one v1's +/// GalaxyAlarmTracker exposed; out-of-range inputs clamp into the nearest bucket. ⚠️ It treats a higher number +/// as more severe, while AVEVA ranks priority 1 as the most urgent: lmxopcua#887 tracks that. Change it here only, so +/// live and history move together. +/// +public static class GalaxyAlarmPriority +{ + /// OPC UA severity of the Low bucket (raw < 250). + public const int OpcUaSeverityLow = 250; + + /// OPC UA severity of the Medium bucket (raw 250–499). + public const int OpcUaSeverityMedium = 500; + + /// OPC UA severity of the High bucket (raw 500–749). + public const int OpcUaSeverityHigh = 750; + + /// OPC UA severity of the Critical bucket (raw 750 and above). + public const int OpcUaSeverityCritical = 1000; + + /// Maps a raw Galaxy priority onto the bucket and its OPC UA severity. + /// The raw value (0–999; clamped). + /// The bucket and its OPC UA Part 9 numeric severity. + public static (AlarmSeverity Bucket, int OpcUaSeverity) Map(int rawPriority) => rawPriority switch + { + < 250 => (AlarmSeverity.Low, OpcUaSeverityLow), + < 500 => (AlarmSeverity.Medium, OpcUaSeverityMedium), + < 750 => (AlarmSeverity.High, OpcUaSeverityHigh), + _ => (AlarmSeverity.Critical, OpcUaSeverityCritical), + }; +} diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/GalaxyAttributeRef.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/GalaxyAttributeRef.cs new file mode 100644 index 000000000..db8a02ed9 --- /dev/null +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/GalaxyAttributeRef.cs @@ -0,0 +1,75 @@ +using System.Text.Json; + +namespace ZB.MOM.WW.OtOpcUa.Core.Abstractions; + +/// +/// The Galaxy () tag reference: the attributeRef a Galaxy raw tag's +/// TagConfig carries, tag_name.AttributeName, which the driver dials through MXAccess. One parser for the +/// driver (which dials it) and the address-space composer (which names the AVEVA event source it implies, Gitea #886), +/// so the two can never disagree about which tags carry one. +/// +public static class GalaxyAttributeRef +{ + /// The TagConfig key (camelCase; matched case-insensitively). + public const string Key = "attributeRef"; + + /// + /// Read the Galaxy dialled address from a tag's TagConfig JSON — the camelCase attributeRef key (the v3 + /// rename of the pre-v3 PascalCase FullName). Case-insensitive on the property name for authoring tolerance; + /// returns null on an absent/blank/non-string value or an unparseable blob. + /// + /// The tag's TagConfig JSON. + /// The attributeRef, or null. + public static string? TryRead(string? tagConfigJson) + { + if (string.IsNullOrWhiteSpace(tagConfigJson)) return null; + try + { + using var doc = JsonDocument.Parse(tagConfigJson); + if (doc.RootElement.ValueKind != JsonValueKind.Object) return null; + foreach (var prop in doc.RootElement.EnumerateObject()) + { + if (!prop.NameEquals(Key) && !string.Equals(prop.Name, Key, StringComparison.OrdinalIgnoreCase)) continue; + if (prop.Value.ValueKind != JsonValueKind.String) return null; + var v = prop.Value.GetString(); + return string.IsNullOrWhiteSpace(v) ? null : v; + } + } + catch (JsonException) + { + // Malformed blob: no dialable address. + } + + return null; + } + + /// + /// The AVEVA alarm reference of a native-alarm tag, or null: the tag's attributeRef when its + /// driver is the Galaxy driver, otherwise null (no other driver's alarms are stored by AVEVA under a Galaxy name). + /// + /// The owning driver's DriverType. + /// The tag's TagConfig JSON. + /// The Galaxy alarm reference (tag_name.AlarmAttribute), or null. + public static string? AlarmReferenceOf(string? driverType, string? tagConfigJson) => + string.Equals(driverType, DriverTypeNames.Galaxy, StringComparison.Ordinal) ? TryRead(tagConfigJson) : null; + + /// + /// The Galaxy object (tag_name) a reference belongs to: the text before the first ., or the whole + /// reference when it has none. This is AVEVA's Source_Object for the object's events and its attributes' + /// alarms. Null for a blank reference or one that starts with a .. + /// + /// A Galaxy reference, e.g. Oven_007.TempHiAlarm. + /// The object name, e.g. Oven_007. + public static string? ObjectNameOf(string? reference) + { + if (string.IsNullOrWhiteSpace(reference)) return null; + var trimmed = reference.Trim(); + var dot = trimmed.IndexOf('.'); + return dot switch + { + < 0 => trimmed, + 0 => null, + _ => trimmed[..dot], + }; + } +} 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 047d5b83d..ff492fd37 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 @@ -128,6 +128,11 @@ public sealed class HistoryReadLimitExceededException : InvalidOperationExceptio /// The stored alarm id (a native alarm's RawPath / a scripted alarm's id) — what the server resolves the event's /// SourceNode (the condition node) from, when that condition is materialised. /// +/// +/// The upstream system's own reference to the alarm condition, when the backend has one — AVEVA's +/// source_conditionvariable (tag_name.AlarmAttribute) on a Galaxy alarm row. The server resolves it to the +/// native condition that dials the same reference, among those that fire on the folder being read (Gitea #886). +/// public sealed record HistoricalEvent( string EventId, string? SourceName, @@ -137,7 +142,8 @@ public sealed record HistoricalEvent( ushort Severity, string? EventTypeName = null, string? ConditionName = null, - string? AlarmId = null); + string? AlarmId = null, + string? ConditionReference = null); /// Result of a call. /// Events ordered by EventTimeUtc in the read's direction — ascending, or descending for a reverse read. diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/GalaxyDriver.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/GalaxyDriver.cs index a670da70a..277f73a0e 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/GalaxyDriver.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/GalaxyDriver.cs @@ -454,31 +454,7 @@ public sealed class GalaxyDriver /// on the property name for authoring tolerance; returns null on an absent/blank/non-string value /// or unparseable blob (the entry is then skipped, never dialled). /// - private static string? TryReadAttributeRef(string tagConfigJson) - { - if (string.IsNullOrWhiteSpace(tagConfigJson)) return null; - try - { - using var doc = JsonDocument.Parse(tagConfigJson); - if (doc.RootElement.ValueKind != JsonValueKind.Object) return null; - foreach (var prop in doc.RootElement.EnumerateObject()) - { - if (!prop.NameEquals("attributeRef") - && !string.Equals(prop.Name, "attributeRef", StringComparison.OrdinalIgnoreCase)) - { - continue; - } - if (prop.Value.ValueKind != JsonValueKind.String) return null; - var v = prop.Value.GetString(); - return string.IsNullOrWhiteSpace(v) ? null : v; - } - } - catch (JsonException) - { - // Malformed blob — treat as un-dialable; BuildRawTagTables logs the skip. - } - return null; - } + private static string? TryReadAttributeRef(string tagConfigJson) => GalaxyAttributeRef.TryRead(tagConfigJson); /// RawPath → Galaxy attributeRef for MXAccess dialling; identity on a miss. private string ToAttributeRef(string rawPath) => diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/MxAccessSeverityMapper.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/MxAccessSeverityMapper.cs index 4c5b05f98..ace46a375 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/MxAccessSeverityMapper.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/MxAccessSeverityMapper.cs @@ -24,13 +24,13 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Runtime; internal static class MxAccessSeverityMapper { /// OPC UA Part 9 numeric severity for the Low bucket (0-249 MxAccess). - public const int OpcUaSeverityLow = 250; + public const int OpcUaSeverityLow = GalaxyAlarmPriority.OpcUaSeverityLow; /// OPC UA Part 9 numeric severity for the Medium bucket (250-499 MxAccess). - public const int OpcUaSeverityMedium = 500; + public const int OpcUaSeverityMedium = GalaxyAlarmPriority.OpcUaSeverityMedium; /// OPC UA Part 9 numeric severity for the High bucket (500-749 MxAccess). - public const int OpcUaSeverityHigh = 750; + public const int OpcUaSeverityHigh = GalaxyAlarmPriority.OpcUaSeverityHigh; /// OPC UA Part 9 numeric severity for the Critical bucket (750+ MxAccess). - public const int OpcUaSeverityCritical = 1000; + public const int OpcUaSeverityCritical = GalaxyAlarmPriority.OpcUaSeverityCritical; /// /// Translate a raw MXAccess severity into the four-bucket @@ -38,20 +38,6 @@ internal static class MxAccessSeverityMapper /// /// The raw MXAccess severity value (0-999 range, clamped if out of range). /// The mapped four-bucket and its OPC UA Part 9 numeric severity. - public static (AlarmSeverity Bucket, int OpcUaSeverity) Map(int rawMxAccessSeverity) - { - if (rawMxAccessSeverity < 250) - { - return (AlarmSeverity.Low, OpcUaSeverityLow); - } - if (rawMxAccessSeverity < 500) - { - return (AlarmSeverity.Medium, OpcUaSeverityMedium); - } - if (rawMxAccessSeverity < 750) - { - return (AlarmSeverity.High, OpcUaSeverityHigh); - } - return (AlarmSeverity.Critical, OpcUaSeverityCritical); - } + // One table for live and history (Gitea #886): GalaxyAlarmPriority, which AVEVA history rows map their priority through. + public static (AlarmSeverity Bucket, int OpcUaSeverity) Map(int rawMxAccessSeverity) => GalaxyAlarmPriority.Map(rawMxAccessSeverity); } diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/Mapping/EventMapper.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/Mapping/EventMapper.cs index facbe5f8f..c9840ac7d 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/Mapping/EventMapper.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/Mapping/EventMapper.cs @@ -8,6 +8,20 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Mapping; /// Maps a gateway wire event () onto the driver-agnostic /// consumed by the Server's HistoryReadEvents path. /// +/// +/// +/// Property names are matched case-insensitively (Gitea #886). AVEVA stores every event property under a lower-case +/// key: its own A&E rows carry priority, alarm_id, source_conditionvariable, and OtOpcUa's own +/// events, written as AlarmName / AlarmId / Severity / Message, come back as +/// alarmname / alarmid / severity / message. +/// +/// +/// Severity: an AVEVA row's priority maps through , the table live Galaxy alarms +/// use, so a row and its live event agree. AVEVA's own severity on such a row (1–4, 0 for a system event) is not on +/// the OPC UA scale and is ignored. A row with no priority (OtOpcUa's own) takes severity, clamped to +/// 1–1000. +/// +/// internal static class EventMapper { /// OPC UA severity range (Part 9): 1 (lowest) … 1000 (highest). @@ -19,13 +33,16 @@ internal static class EventMapper /// The driver-agnostic historical event. public static HistoricalEvent ToHistoricalEvent(HistorianEvent historianEvent) { + var properties = historianEvent.Properties; + // Message: prefer the "Message" property, else fall back to the event Type (best-effort // render); never null-crash on a missing property. - string? message; - if (historianEvent.Properties.TryGetValue("Message", out var m) && !string.IsNullOrEmpty(m)) - message = m; - else - message = string.IsNullOrEmpty(historianEvent.Type) ? null : historianEvent.Type; + var message = Property(properties, "Message") + ?? (string.IsNullOrEmpty(historianEvent.Type) ? null : historianEvent.Type); + + // AVEVA's reference to the alarm condition (tag_name.AlarmAttribute); the server resolves it to the native + // condition that dials it, when one fires on the folder being read. + var conditionReference = Property(properties, "source_conditionvariable"); return new HistoricalEvent( EventId: historianEvent.Id, @@ -33,15 +50,32 @@ internal static class EventMapper EventTimeUtc: historianEvent.EventTime?.ToDateTime() ?? default, // Utc kind ReceivedTimeUtc: historianEvent.ReceivedTime?.ToDateTime() ?? default, // Utc kind Message: message, - Severity: ParseSeverity(historianEvent.Properties), + Severity: ParseSeverity(properties), // What GatewayAlarmHistorianWriter sent (AlarmEventMapper): Type = AlarmTypeName, AlarmName / AlarmId properties. EventTypeName: string.IsNullOrEmpty(historianEvent.Type) ? null : historianEvent.Type, - ConditionName: Property(historianEvent.Properties, "AlarmName"), - AlarmId: Property(historianEvent.Properties, "AlarmId")); + ConditionName: Property(properties, "AlarmName") ?? AttributeOf(conditionReference), + AlarmId: Property(properties, "AlarmId") ?? Property(properties, "alarm_id"), + ConditionReference: conditionReference); } - private static string? Property(IDictionary properties, string name) => - properties.TryGetValue(name, out var v) && !string.IsNullOrEmpty(v) ? v : null; + /// A property by name, exact first, else case-insensitively; null when absent or empty. + private static string? Property(IDictionary properties, string name) + { + if (properties.TryGetValue(name, out var exact)) return string.IsNullOrEmpty(exact) ? null : exact; + foreach (var (key, value) in properties) + if (string.Equals(key, name, StringComparison.OrdinalIgnoreCase)) + return string.IsNullOrEmpty(value) ? null : value; + return null; + } + + /// The attribute part of a Galaxy reference (Oven_007.TempHiAlarm ⇒ TempHiAlarm), or the whole + /// reference when it has no object prefix. + private static string? AttributeOf(string? reference) + { + if (reference is null) return null; + var dot = reference.IndexOf('.'); + return dot > 0 && dot < reference.Length - 1 ? reference[(dot + 1)..] : reference; + } /// Maps a batch of gateway events to historical events, in order. /// The gateway wire events. @@ -55,20 +89,20 @@ internal static class EventMapper } /// - /// Parses an OPC UA severity from the "Severity" property (else "Priority"), clamped to - /// [1, 1000]. Missing or unparseable values default to the minimum severity (1). + /// The OPC UA severity: an AVEVA row's priority through the live Galaxy table, else the Severity property + /// clamped to [1, 1000]. Missing or unparseable values default to the minimum severity (1). /// private static ushort ParseSeverity(IDictionary properties) { - string? raw = null; - if (properties.TryGetValue("Severity", out var severity)) - raw = severity; - else if (properties.TryGetValue("Priority", out var priority)) - raw = priority; + if (TryParseInt(Property(properties, "Priority"), out var priority)) + return (ushort)GalaxyAlarmPriority.Map(priority).OpcUaSeverity; - if (int.TryParse(raw, NumberStyles.Integer, CultureInfo.InvariantCulture, out var value)) + if (TryParseInt(Property(properties, "Severity"), out var value)) return (ushort)Math.Clamp(value, MinSeverity, MaxSeverity); return MinSeverity; } + + private static bool TryParseInt(string? raw, out int value) => + int.TryParse(raw, NumberStyles.Integer, CultureInfo.InvariantCulture, out value); } diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/AddressSpaceApplier.cs b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/AddressSpaceApplier.cs index 961319f5b..518a54548 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/AddressSpaceApplier.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/AddressSpaceApplier.cs @@ -918,8 +918,15 @@ public sealed class AddressSpaceApplier if (!SafeMaterialiseAlarmCondition(t.NodeId, t.ParentNodeId ?? string.Empty, t.Name, t.Alarm.AlarmType, t.Alarm.Severity, isNative: true, AddressSpaceRealm.Raw)) { failed++; + continue; } - else if (t.ReferencingEquipmentPaths.Count > 0) + + // #886: a Galaxy alarm's AVEVA reference, so every folder this condition fires on also reads the + // history AVEVA stores under its object. Bound before the wiring; the folders are read at HistoryRead. + if (t.NativeAlarmReference is { } nativeReference && !SafeBindNativeAlarmHistory(t.NodeId, nativeReference)) + failed++; + + if (t.ReferencingEquipmentPaths.Count > 0) { // Multi-notifier fan-out: wire the SINGLE condition as an event notifier of each // referencing equipment's UNS folder, so one ReportEvent reaches every referencing equipment @@ -1247,6 +1254,12 @@ public sealed class AddressSpaceApplier /// swallowing (and Warning-logging) any sink fault. Returns true on success, false when the /// sink threw — callers tally the false into their pass's failed-node count (archreview 01/S-1). /// true when the notifiers were wired; false when the sink threw. + private bool SafeBindNativeAlarmHistory(string alarmNodeId, string alarmReference) + { + try { _sink.BindNativeAlarmHistory(alarmNodeId, AddressSpaceRealm.Raw, alarmReference); return true; } + catch (Exception ex) { _logger.LogWarning(ex, "AddressSpaceApplier: BindNativeAlarmHistory threw for {Node}", alarmNodeId); return false; } + } + private bool SafeWireAlarmNotifiers(string alarmNodeId, AddressSpaceRealm alarmRealm, IReadOnlyList notifierFolderNodeIds, AddressSpaceRealm notifierFolderRealm) { try { _sink.WireAlarmNotifiers(alarmNodeId, alarmRealm, notifierFolderNodeIds, notifierFolderRealm); return true; } diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/AddressSpaceComposer.cs b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/AddressSpaceComposer.cs index 44fced51d..84ccc29bc 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/AddressSpaceComposer.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/AddressSpaceComposer.cs @@ -150,6 +150,11 @@ public sealed record RawContainerNode( /// the DRIVER, deliberately not gated on , so a surgical scalar↔array edit does not /// flip it. It is not in the classifier's surgical whitelist, so a change to it (a driver's type edited in /// place) rebuilds. +/// is the Galaxy alarm reference (tag_name.AlarmAttribute, the tag's +/// attributeRef) of a tag that carries an , and null for +/// every other tag. AVEVA stores the alarm's history under that reference's object, so the node manager adds the +/// object to the event-history sources of every folder the condition fires on (Gitea #886). It is not in the +/// surgical whitelist either: an alarm-bearing tag's delta rebuilds anyway. /// ADDING A FIELD IS A FOUR-PLACE CONTRACT — the hand-written equality below does NOT close /// over new members. Because forces a custom /// Equals/GetHashCode, the compiler's positional equality is suppressed, so a field added @@ -178,7 +183,8 @@ public sealed record RawTagPlan( uint? ArrayLength = null, AddressSpaceRealm Realm = AddressSpaceRealm.Raw, bool ArrayWritable = false, - bool ScalarRefusesArrayValue = false) + bool ScalarRefusesArrayValue = false, + string? NativeAlarmReference = null) { /// // Structural equality: the auto-generated record equality compares ReferencingEquipmentPaths (an @@ -202,6 +208,7 @@ public sealed record RawTagPlan( Realm == other.Realm && ArrayWritable == other.ArrayWritable && ScalarRefusesArrayValue == other.ScalarRefusesArrayValue && + NativeAlarmReference == other.NativeAlarmReference && ReferencingEquipmentPaths.SequenceEqual(other.ReferencingEquipmentPaths, StringComparer.Ordinal); /// @@ -211,7 +218,7 @@ public sealed record RawTagPlan( hash.Add(TagId); hash.Add(NodeId); hash.Add(ParentNodeId); hash.Add(DriverInstanceId); hash.Add(Name); hash.Add(DataType); hash.Add(Writable); hash.Add(Alarm); hash.Add(IsHistorized); hash.Add(HistorianTagname); hash.Add(IsArray); hash.Add(ArrayLength); - hash.Add(Realm); hash.Add(ArrayWritable); hash.Add(ScalarRefusesArrayValue); + hash.Add(Realm); hash.Add(ArrayWritable); hash.Add(ScalarRefusesArrayValue); hash.Add(NativeAlarmReference); foreach (var p in ReferencingEquipmentPaths) hash.Add(p, StringComparer.Ordinal); return hash.ToHashCode(); } @@ -803,6 +810,7 @@ public static class AddressSpaceComposer var equipPathByTag = BuildReferencingEquipmentPathsByTag(unsTagReferences, unsAreas, unsLines, equipment); var arrayWritableFor = BuildArrayWriteCapability(driverInstances, devices); var scalarRefusesArrayFor = BuildScalarArrayRefusal(driverInstances, devices); + var nativeAlarmReferenceFor = BuildNativeAlarmReference(driverInstances, devices); var rawTags = new List(); foreach (var t in tags ?? Array.Empty()) @@ -820,6 +828,8 @@ public static class AddressSpaceComposer ? new EquipmentTagAlarmInfo(a.AlarmType, a.Severity, a.HistorizeToAveva) : null; var refEquipPaths = equipPathByTag.GetValueOrDefault(t.TagId, Array.Empty()); + // #886: the AVEVA history source of a native Galaxy alarm (null for every other tag). + var nativeAlarmReference = alarm is null ? null : nativeAlarmReferenceFor(t); rawTags.Add(new RawTagPlan( TagId: t.TagId, @@ -836,7 +846,8 @@ public static class AddressSpaceComposer IsArray: intent.IsArray, ArrayLength: intent.ArrayLength, ArrayWritable: arrayWritable, - ScalarRefusesArrayValue: scalarRefusesArray)); + ScalarRefusesArrayValue: scalarRefusesArray, + NativeAlarmReference: nativeAlarmReference)); } containers.Sort((x, y) => string.CompareOrdinal(x.NodeId, y.NodeId)); @@ -876,6 +887,17 @@ public static class AddressSpaceComposer return tag => DriverArrayWriteSupport.RefusesArrayValueOnScalarTag(resolve(tag).Driver?.DriverType); } + /// + /// A tag's Galaxy alarm reference (its attributeRef) when its driver is the Galaxy driver, else null — read + /// with the driver's own parser (), so a tag the driver cannot dial names no source. + /// + private static Func BuildNativeAlarmReference( + IReadOnlyList driverInstances, IReadOnlyList? devices) + { + var resolve = BuildTagOwnerResolver(driverInstances, devices); + return tag => GalaxyAttributeRef.AlarmReferenceOf(resolve(tag).Driver?.DriverType, tag.TagConfig); + } + // Tag → (owning driver, owning device); first row wins on a duplicate id, matching the composer's other id maps. private static Func BuildTagOwnerResolver( IReadOnlyList driverInstances, IReadOnlyList? devices) diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/EventSourceFanOut.cs b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/EventSourceFanOut.cs new file mode 100644 index 000000000..58b56c0a0 --- /dev/null +++ b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/EventSourceFanOut.cs @@ -0,0 +1,193 @@ +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; + +namespace ZB.MOM.WW.OtOpcUa.OpcUaServer; + +/// A native condition an AVEVA history row can resolve to (Gitea #886). +/// The condition's id (its RawPath), from which the server resolves the row's SourceNode. +/// The live condition's ConditionName. +internal sealed record EventHistoryCondition(string AlarmId, string ConditionName); + +/// +/// What one notifier folder's event history reads (Gitea #886): its event sources, and the native conditions that fire +/// on it live keyed by their Galaxy alarm reference. +/// +internal sealed class EventHistoryScope +{ + private static readonly IReadOnlyDictionary NoConditions = + new Dictionary(); + + private readonly IReadOnlyDictionary _conditions; + + private EventHistoryScope(IReadOnlyList sources, IReadOnlyDictionary conditions) + { + Sources = sources; + _conditions = conditions; + // One source keeps its bare name, so a pre-#886 folder's cursor is exactly what it was. + CursorKey = string.Join('\n', sources); + } + + /// The sources to read: distinct, ordinal-sorted (the folder's own first when it is the only one). + public IReadOnlyList Sources { get; } + + /// The identity a continuation point is bound to: the whole source set. + public string CursorKey { get; } + + /// A scope of one source and no Galaxy conditions (every non-Galaxy folder). + /// The folder's own source. + /// The scope. + public static EventHistoryScope Single(string source) => new([source], NoConditions); + + /// A folder's own source plus the Galaxy objects behind it. + /// The folder's own source. + /// The Galaxy objects of the native conditions that fire on it (duplicates allowed). + /// Those conditions by Galaxy alarm reference (case-insensitive). + /// The scope. + public static EventHistoryScope Create( + string ownSource, IEnumerable? objects, IReadOnlyDictionary conditions) + { + var sources = new SortedSet(StringComparer.Ordinal) { ownSource }; + foreach (var o in objects ?? []) sources.Add(o); + return new EventHistoryScope(sources.ToList(), conditions); + } + + /// + /// Give each row whose names one of this folder's native conditions + /// that condition's id (so its SourceNode resolves) and live ConditionName. Other rows are unchanged. + /// + /// A page of events. + /// The page, resolved. + public IReadOnlyList ResolveConditions(IReadOnlyList events) + { + if (_conditions.Count == 0 || events.Count == 0) return events; + List? resolved = null; + for (var i = 0; i < events.Count; i++) + { + var e = events[i]; + if (e.ConditionReference is not { } reference || !_conditions.TryGetValue(reference, out var condition)) continue; + resolved ??= [.. events]; + resolved[i] = e with { AlarmId = condition.AlarmId, ConditionName = condition.ConditionName }; + } + + return resolved ?? events; + } +} + +/// +/// Reads several event sources as one (Gitea #886): one backend read per source (the historiangw ReadEvents +/// filter is a single predicate, so there is no multi-source request), merged, de-duplicated by EventId and cut to the +/// cap. +/// +/// +/// +/// Merge order is (EventTimeUtc, source, arrival): time in the read's direction, then the source's +/// position in the folder's (ordinal-sorted) source list — reversed for a reverse read — then the order the source +/// returned its events in. A source's own order within a timestamp is all the paging needs to be stable (the +/// Gateway orders by event time only), so ties are never re-sorted by EventId here: re-sorting could move an +/// unread tie in front of one already sent. The reverse form is the forward form backwards, which is what the +/// server's tie-cluster over-fetch (a forward read, reversed) relies on. A walked-back reverse read (Gateway) sorts +/// its ties by EventId itself, as before. +/// +/// +/// Truncation stays honest. A source that flags "more" returned a prefix of its own stream, ending at time H. +/// Its unread events lie at H or later (in the read's direction), and at H they follow what it returned. So the +/// page keeps every event before the earliest such H, then, at H, each source's events in merge order up to and +/// including the first source truncated at H — never an event the unread ones would precede. The page is never +/// empty, because that source returned at least one event at H. When every source is read with the same cap (the +/// server always passes one) a page cut to that cap already stops there; the bound matters for an uncapped or +/// default-capped read. +/// +/// A single source is passed straight through, so a folder with one source reads exactly as before #886. +/// +internal static class EventSourceFanOut +{ + /// Reads over one window and merges them. + /// The backend. + /// The event sources (at least one). + /// The window start (the later bound for a reverse read). + /// The window end. + /// The cap, passed to every source; <= 0 is the backend's default cap. + /// Newest first. + /// Cancellation. + /// The merged events, and a non-null continuation point when more exist. + public static async Task ReadAsync( + IHistorianDataSource source, IReadOnlyList sources, DateTime startUtc, DateTime endUtc, int maxEvents, + bool reverse, CancellationToken ct) + { + if (sources.Count == 1) + return await source.ReadEventsAsync(sources[0], startUtc, endUtc, maxEvents, reverse, ct).ConfigureAwait(false); + + var merged = new List(); + var seen = new HashSet(StringComparer.Ordinal); + var more = false; + // Per source: the time its truncated read ended at, or null when it returned everything it holds. + var truncatedAt = new DateTime?[sources.Count]; + for (var s = 0; s < sources.Count; s++) + { + // Sequential: the folder's sources share the backend's session pool, and the batch already reads folders in + // parallel. + var read = await source.ReadEventsAsync(sources[s], startUtc, endUtc, maxEvents, reverse, ct).ConfigureAwait(false); + for (var i = 0; i < read.Events.Count; i++) + { + var e = read.Events[i]; + if (string.IsNullOrEmpty(e.EventId) || seen.Add(e.EventId)) merged.Add(new Merged(e, s, i)); + } + + if (read.ContinuationPoint is null) continue; + more = true; + // A truncated read with no events gives no bound; the cap below still flags the page incomplete. + if (read.Events.Count > 0) truncatedAt[s] = read.Events[^1].EventTimeUtc; + } + + var order = new MergeOrder(reverse); + merged.Sort(order); + + // H: the earliest (in the read's direction) time a truncated source stopped at. + DateTime? horizon = null; + foreach (var t in truncatedAt) + if (t is { } at && (horizon is not { } h || (reverse ? at > h : at < h))) horizon = at; + + var page = new List(merged.Count); + foreach (var m in merged) + { + if (horizon is { } h) + { + var t = m.Event.EventTimeUtc; + if (reverse ? t < h : t > h) break; + // At H, merge order visits whole sources in turn: stop after the first one that was cut there. + if (t == h && page.Count > 0 && page[^1].EventTimeUtc == h && StoppedAt(merged, page.Count, h)) break; + } + + page.Add(m.Event); + } + + if (maxEvents > 0 && page.Count > maxEvents) + { + page.RemoveRange(maxEvents, page.Count - maxEvents); + more = true; + } + + return new HistoricalEventsResult(page, more ? [0x01] : null); + + // Whether the event before merged[next] closed a source that was truncated at h (and merged[next] is another source's). + bool StoppedAt(List all, int next, DateTime h) + { + var previous = all[next - 1]; + return previous.Source != all[next].Source && truncatedAt[previous.Source] == h; + } + } + + /// One merged event: the event, its source's position in the list, and its position in that source's read. + private readonly record struct Merged(HistoricalEvent Event, int Source, int Arrival); + + /// Time in the read's direction, then source position (reversed for a reverse read), then arrival. + private sealed class MergeOrder(bool reverse) : IComparer + { + public int Compare(Merged x, Merged y) + { + var c = x.Event.EventTimeUtc.CompareTo(y.Event.EventTimeUtc); + if (c == 0) c = x.Source.CompareTo(y.Source); + if (reverse) c = -c; + return c != 0 ? c : x.Arrival.CompareTo(y.Arrival); + } + } +} diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs index 142786423..922680aee 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs @@ -82,6 +82,13 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3 /// so a re-materialise as the other kind is correct. private readonly HashSet _nativeAlarmNodeIds = new(); /// + /// Gitea #886 — native condition key → its Galaxy alarm reference (tag_name.AlarmAttribute), recorded by + /// . Every folder the condition fires on live reads the history AVEVA stores + /// under the reference's object (). Guarded by Lock; maintained in + /// lock-step with (dropped wherever a condition is, cleared on rebuild). + /// + private readonly Dictionary _nativeAlarmReferences = new(); + /// /// Gitea #797 — the explicit binding from each LIVE scripted condition to the equipment that owns it. /// A scripted condition's NodeId is its opaque ScriptedAlarmId and cannot be split into an ACL /// scope, so with Security:Acl:Enabled its Part 9 commands are authorized at the owner's scope @@ -2319,6 +2326,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3 // H6a: re-materialising the same id as the OTHER kind (native↔scripted) must reflect the new // kind, so always drop the stale native flag first and only re-add it below when isNative. _nativeAlarmNodeIds.Remove(conditionKey); + _nativeAlarmReferences.Remove(conditionKey); // Gitea #797: the dropped instance's owner binding goes with it; the new one is bound below. _scriptedAlarmOwners.Remove(conditionKey); @@ -2721,6 +2729,21 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3 } } + /// + public void BindNativeAlarmHistory(string alarmNodeId, AddressSpaceRealm alarmRealm, string alarmReference) + { + ArgumentException.ThrowIfNullOrEmpty(alarmNodeId); + if (string.IsNullOrWhiteSpace(alarmReference)) return; + var key = MapKey(alarmRealm, alarmNodeId); + lock (Lock) + { + // Only a materialised NATIVE condition carries an upstream reference; anything else (a mid-rebuild race, a + // scripted id) is a no-op, never a throw — a deploy must not fault on it. + if (!_alarmConditions.ContainsKey(key) || !_nativeAlarmNodeIds.Contains(key)) return; + _nativeAlarmReferences[key] = alarmReference.Trim(); + } + } + /// Tear down (bidirectionally) every notifier wired for the condition at /// and drop its tracking entry. MUST be called under Lock. Used on condition-removal + full rebuild /// so no inverse-notifier entry leaks. A no-op when the condition has no wired notifiers. @@ -4648,6 +4671,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3 // H6a: drop the native-alarm flags in lock-step with the conditions they classify, so a // re-materialise on the next apply (possibly as the other kind) starts from a clean slate. _nativeAlarmNodeIds.Clear(); + _nativeAlarmReferences.Clear(); _scriptedAlarmOwners.Clear(); foreach (var f in _folders.Values) PredefinedNodes?.Remove(f.NodeId); @@ -4812,6 +4836,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3 condition.Parent?.RemoveChild(condition); ForgetPredefinedSubtree(condition); _nativeAlarmNodeIds.Remove(key); + _nativeAlarmReferences.Remove(key); _scriptedAlarmOwners.Remove(key); LowerAllEventsMonitoring(condition); e = BuildNodesRemovedModelChange(alarmNodeId, realm); @@ -4922,6 +4947,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3 if (c.Parent is { } parent && !removedParents.Contains(parent)) parent.RemoveChild(c); ForgetPredefinedSubtree(c); _nativeAlarmNodeIds.Remove(key); + _nativeAlarmReferences.Remove(key); _scriptedAlarmOwners.Remove(key); LowerAllEventsMonitoring(c); return; @@ -7830,8 +7856,10 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3 continue; } - var sn = sourceName; - work.Add(() => ServeEventsAsync(handle, sn, details, selectClauses, where, filterContext, nodesToRead, session, + // #886: the folder's own source plus the Galaxy objects of the native conditions that fire on it, taken + // under Lock now so the whole read (every page source, every SourceNode) sees one membership. + var scope = ResolveEventHistoryScope(handle.NodeId, sourceName); + work.Add(() => ServeEventsAsync(handle, scope, details, selectClauses, where, filterContext, nodesToRead, session, results, errors, ct)); } @@ -7839,6 +7867,42 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3 }); } + /// + /// Gitea #886 — what one notifier folder's event history reads: its own registered source (OtOpcUa's own events, + /// as before) plus the Galaxy object of every native condition that fires on it live — those parented under it + /// and those wired to it — and those conditions by their Galaxy reference, so an + /// AVEVA row resolves to the same condition its live event comes from. Taken under Lock. + /// + /// The notifier folder. + /// The folder's registered event source (its NodeId identifier). + /// The folder's scope; just when no Galaxy condition fires on it. + private EventHistoryScope ResolveEventHistoryScope(NodeId folderNodeId, string ownSource) + { + lock (Lock) + { + if (_nativeAlarmReferences.Count == 0 || !_notifierFolders.TryGetValue(folderNodeId, out var folder)) + return EventHistoryScope.Single(ownSource); + + List? objects = null; + Dictionary? conditions = null; + foreach (var (key, reference) in _nativeAlarmReferences) + { + if (!_alarmConditions.TryGetValue(key, out var alarm)) continue; + var firesHere = ReferenceEquals(alarm.Parent, folder) + || (_alarmNotifierWiring.TryGetValue(key, out var wiring) + && ReferenceEquals(wiring.Alarm, alarm) && wiring.Folders.Contains(folder)); + if (!firesHere) continue; + if (GalaxyAttributeRef.ObjectNameOf(reference) is { } objectName) (objects ??= []).Add(objectName); + // Galaxy names are case-insensitive, and the driver matches its live alarm references the same way. + conditions ??= new Dictionary(StringComparer.OrdinalIgnoreCase); + conditions.TryAdd(reference, new EventHistoryCondition( + key.Id, alarm.ConditionName?.Value ?? alarm.DisplayName.Text ?? key.Id)); + } + + return conditions is null ? EventHistoryScope.Single(ownSource) : EventHistoryScope.Create(ownSource, objects, conditions); + } + } + /// /// Serve one registered event-history notifier handle: reads the source's events over the request /// window, projects them onto the select clauses, and fills the service-level results/errors slots. @@ -7847,7 +7911,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3 /// BadTimeout. /// /// The notifier node handle; handle.Index indexes results/errors. - /// The registered historian event-source name for this notifier. + /// The notifier's event-history sources and the native conditions that fire on it (#886). /// The event-read details (window + count cap). /// The event-filter select clauses (the fields to emit, in order). /// The event filter's where-clause, or null for none. @@ -7857,7 +7921,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3 /// Per-request deadline token. private async Task ServeEventsAsync( NodeHandle handle, - string? sourceName, + EventHistoryScope scope, ReadEventDetails details, ArrayOf selectClauses, ContentFilter? where, @@ -7873,6 +7937,12 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3 var resumed = inboundCp is { Length: > 0 }; // One snapshot of the (hot-swappable) source for the whole read: the capability check and every call must agree. var source = _historianDataSource; + // The folder's sources read as ONE source (#886): a single source is read directly, several are fanned out and + // merged in (EventTimeUtc, source, arrival) order (EventSourceFanOut), so the paging below is the same either way. + // The cursor is bound to the whole source set. + var sourceName = scope.CursorKey; + Task ReadSourcesAsync(DateTime from, DateTime to, int max, bool rev, CancellationToken token) => + EventSourceFanOut.ReadAsync(source, scope.Sources, from, to, max, rev, token); try { var startUtc = SdkValueBoundary.ToClrTimestamp(details.StartTime); @@ -7888,14 +7958,13 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3 // for THIS node only. var state = _historyContinuationStore.TryTake(session, inboundCp.ToArray()); // A point is bound to the node that issued it: another node's source name is a miss too. - if (state is null || !HistoryContinuationMatches(state, handle.NodeId, historyEpoch) || state.Kind != HistoryCursorKind.Events || state.Tagname != (sourceName ?? string.Empty)) + if (state is null || !HistoryContinuationMatches(state, handle.NodeId, historyEpoch) || state.Kind != HistoryCursorKind.Events || state.Tagname != sourceName) { errors[handle.Index] = StatusCodes.BadContinuationPointInvalid; results[handle.Index] = new SdkHistoryReadResult { StatusCode = StatusCodes.BadContinuationPointInvalid }; return; } - sourceName = state.Tagname; startUtc = state.NextStartUtc; endUtc = state.EndUtc; boundarySkip = state.BoundarySkipCount; @@ -7956,8 +8025,8 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3 // continuation point (the walk itself cannot flag truncation). var want = numValuesPerNode == 0 ? 0u : (uint)Math.Min(uint.MaxValue - 1L, (long)fetch + 1); var (rows, next) = await ReverseEventsReader.ReadAsync( - source, sourceName, startUtc, endUtc, want, MaxTieClusterOverfetch, Clock.GetUtcNow().UtcDateTime, - walkHint, ct) + (from, to, cap, token) => ReadSourcesAsync(from, to, cap, rev: false, token), source.TimestampResolutionTicks, + sourceName, startUtc, endUtc, want, MaxTieClusterOverfetch, Clock.GetUtcNow().UtcDateTime, walkHint, ct) .ConfigureAwait(false); walkHint = next; var more = want > 0 && rows.Count > fetch; @@ -7966,8 +8035,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3 } else { - sourceResult = await source.ReadEventsAsync( - sourceName, startUtc, endUtc, fetch, reverse, ct).ConfigureAwait(false); + sourceResult = await ReadSourcesAsync(startUtc, endUtc, fetch, reverse, ct).ConfigureAwait(false); } // Truncation is the SOURCE's word (non-null continuation point) — never inferred from the count, so a @@ -8004,8 +8072,8 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3 // with an explicit bounded cap and page WITHIN it (same scheme as raw). var resolution = Math.Max(1L, source.TimestampResolutionTicks); var overfetchCap = (int)Math.Min(int.MaxValue - 1L, (long)Math.Max(1, MaxTieClusterOverfetch) + 1); - IReadOnlyList cluster = (await source.ReadEventsAsync( - sourceName, startUtc, startUtc.AddTicks(resolution), overfetchCap, reverse: false, ct) + IReadOnlyList cluster = (await ReadSourcesAsync( + startUtc, startUtc.AddTicks(resolution), overfetchCap, rev: false, ct) .ConfigureAwait(false)).Events; // A walked read emitted its ties sorted by EventId; the gateway returns them in any order and its @@ -8041,7 +8109,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3 if (nextStartUtc is { } resumeAt) { pendingCursor = new HistoryContinuationState( - sourceName ?? string.Empty, resumeAt, endUtc, nextSkip, numValuesPerNode, HistoryCursorKind.Events, + sourceName, resumeAt, endUtc, nextSkip, numValuesPerNode, HistoryCursorKind.Events, Reverse: reverse, SegmentWidthTicks: walkHint.WidthTicks, EmptyFromTicks: walkHint.EmptyFromTicks, EmptyToTicks: walkHint.EmptyToTicks); } @@ -8054,7 +8122,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3 var skip = 0; for (var i = events.Count - 1; i >= 0 && events[i].EventTimeUtc == boundary; i--) skip++; pendingCursor = new HistoryContinuationState( - sourceName ?? string.Empty, boundary, endUtc, skip, numValuesPerNode, HistoryCursorKind.Events, + sourceName, boundary, endUtc, skip, numValuesPerNode, HistoryCursorKind.Events, Reverse: reverse, SegmentWidthTicks: walkHint.WidthTicks, EmptyFromTicks: walkHint.EmptyFromTicks, EmptyToTicks: walkHint.EmptyToTicks); } @@ -8062,6 +8130,9 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3 // The where-clause filters AFTER the page's cursor was taken from the unfiltered events, so paging (time, // boundary skip) is exactly as without a filter: a page may come back with fewer events — even none — and a // continuation point, never skipping or repeating one. + // #886: an AVEVA row naming a native condition that fires on this folder carries that condition's SourceNode and + // ConditionName, as the live event does. Rewritten after paging: EventId and time are untouched. + events = scope.ResolveConditions(events); var historyEvent = ProjectEvents(events, selectClauses, where, filterContext); var outboundCp = pendingCursor is null ? null : SaveHistoryContinuation(session, handle.NodeId, historyEpoch, pendingCursor); results[handle.Index] = new SdkHistoryReadResult diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/ReverseEventsReader.cs b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/ReverseEventsReader.cs index db295d1a4..7d714ffa5 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/ReverseEventsReader.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/ReverseEventsReader.cs @@ -27,9 +27,33 @@ internal static class ReverseEventsReader /// Cancellation. /// The events, newest first, and the hint for the next page. /// One timestamp holds more than the tie budget. - public static async Task ReadAsync( + public static Task ReadAsync( IHistorianDataSource source, string? sourceName, DateTime hi, DateTime lo, uint want, int maxTieCluster, - DateTime nowUtc, ReverseWalkHint hint, CancellationToken ct) + DateTime nowUtc, ReverseWalkHint hint, CancellationToken ct) => + ReadAsync( + (start, end, cap, token) => source.ReadEventsAsync(sourceName, start, end, cap, reverse: false, token), + source.TimestampResolutionTicks, sourceName, hi, lo, want, maxTieCluster, nowUtc, hint, ct); + + /// + /// Reads (lo, hi] newest-first from a forward event reader — the form a folder with several event sources + /// uses (, Gitea #886), whose merged reads stand in for one source's. + /// + /// Reads [start, end] forward with a cap (int.MaxValue = none). + /// The backend's timestamp resolution. + /// The source (or source set) named in a limit message. + /// The later (inclusive) bound. + /// The earlier (exclusive) bound. + /// How many events; 0 = all in the window (one forward read). + /// The tie budget (MaxTieClusterOverfetch) — a ceiling only. + /// Now, for the look-ahead clamp. + /// What the previous page carried (default = none). + /// Cancellation. + /// The events, newest first, and the hint for the next page. + /// One timestamp holds more than the tie budget. + public static async Task ReadAsync( + Func> readForward, long resolutionTicks, + string? sourceName, DateTime hi, DateTime lo, uint want, int maxTieCluster, DateTime nowUtc, ReverseWalkHint hint, + CancellationToken ct) { var budget = Math.Max(1, maxTieCluster); var (rows, next) = await ReverseWalk.WalkAsync( @@ -37,11 +61,10 @@ internal static class ReverseEventsReader { // cap 0 = the whole window: int.MaxValue is the events contract's "no cap" (a non-positive value would // mean the backend's DEFAULT cap). - var r = await source.ReadEventsAsync(sourceName, start, end, cap == 0 ? int.MaxValue : cap, reverse: false, token) - .ConfigureAwait(false); + var r = await readForward(start, end, cap == 0 ? int.MaxValue : cap, token).ConfigureAwait(false); return new ForwardSegmentRead(r.Events, r.ContinuationPoint is not null); }, - ev => ev.EventTimeUtc.Ticks, EventIdTieOrder.Instance, hi, lo, want, budget, source.TimestampResolutionTicks, + ev => ev.EventTimeUtc.Ticks, EventIdTieOrder.Instance, hi, lo, want, budget, resolutionTicks, nowUtc, hint, at => new HistoryReadLimitExceededException(budget, $"A reverse events read of '{sourceName}' at {at:O} found more than {budget} events within one millisecond; the " + diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/SdkAddressSpaceSink.cs b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/SdkAddressSpaceSink.cs index 8e3a9f0d3..c939d9363 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/SdkAddressSpaceSink.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/SdkAddressSpaceSink.cs @@ -54,6 +54,10 @@ public sealed class SdkAddressSpaceSink : IOpcUaAddressSpaceSink, ISurgicalAddre public void WireAlarmNotifiers(string alarmNodeId, AddressSpaceRealm alarmRealm, IReadOnlyList notifierFolderNodeIds, AddressSpaceRealm notifierFolderRealm) => _nodeManager.WireAlarmNotifiers(alarmNodeId, alarmRealm, notifierFolderNodeIds, notifierFolderRealm); + /// + public void BindNativeAlarmHistory(string alarmNodeId, AddressSpaceRealm alarmRealm, string alarmReference) + => _nodeManager.BindNativeAlarmHistory(alarmNodeId, alarmRealm, alarmReference); + /// public void EnsureFolder(string folderNodeId, string? parentNodeId, string displayName, AddressSpaceRealm realm) => _nodeManager.EnsureFolder(folderNodeId, parentNodeId, displayName, realm); diff --git a/tests/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions.Tests/GalaxyAttributeRefTests.cs b/tests/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions.Tests/GalaxyAttributeRefTests.cs new file mode 100644 index 000000000..df94e903f --- /dev/null +++ b/tests/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions.Tests/GalaxyAttributeRefTests.cs @@ -0,0 +1,40 @@ +using Shouldly; +using Xunit; + +namespace ZB.MOM.WW.OtOpcUa.Core.Abstractions.Tests; + +/// Gitea #886 — the Galaxy reference parser the driver and the composer share, and the object it names. +public sealed class GalaxyAttributeRefTests +{ + [Theory] + [InlineData("""{"attributeRef":"Oven_007.TempHiAlarm"}""", "Oven_007.TempHiAlarm")] + [InlineData("""{"AttributeRef":"Oven_007.Temp","alarm":{"alarmType":"LimitAlarm"}}""", "Oven_007.Temp")] + [InlineData("""{"attributeRef":" "}""", null)] + [InlineData("""{"attributeRef":7}""", null)] + [InlineData("""{"FullName":"Oven_007.Temp"}""", null)] + [InlineData("""not json""", null)] + [InlineData("", null)] + [InlineData(null, null)] + public void Reads_the_attributeRef(string? tagConfig, string? expected) => + GalaxyAttributeRef.TryRead(tagConfig).ShouldBe(expected); + + [Fact] + public void Only_a_galaxy_tag_names_an_aveva_alarm_reference() + { + const string config = """{"attributeRef":"Oven_007.TempHiAlarm"}"""; + GalaxyAttributeRef.AlarmReferenceOf(DriverTypeNames.Galaxy, config).ShouldBe("Oven_007.TempHiAlarm"); + GalaxyAttributeRef.AlarmReferenceOf(DriverTypeNames.Modbus, config).ShouldBeNull(); + GalaxyAttributeRef.AlarmReferenceOf(null, config).ShouldBeNull(); + } + + [Theory] + [InlineData("Oven_007.TempHiAlarm", "Oven_007")] + [InlineData("Oven_007.Temp.HiHi", "Oven_007")] + [InlineData("Oven_007", "Oven_007")] + [InlineData(" Oven_007.Temp ", "Oven_007")] + [InlineData(".Temp", null)] + [InlineData("", null)] + [InlineData(null, null)] + public void The_object_is_the_reference_up_to_its_first_dot(string? reference, string? expected) => + GalaxyAttributeRef.ObjectNameOf(reference).ShouldBe(expected); +} diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/Mapping/EventMapperTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/Mapping/EventMapperTests.cs index b46e44b76..7b46bcd59 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/Mapping/EventMapperTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/Mapping/EventMapperTests.cs @@ -43,7 +43,7 @@ public sealed class EventMapperTests } [Theory] - [InlineData("Priority", "999", 999)] + [InlineData("Priority", "999", 1000)] // a Galaxy priority maps through the live table (#886), not clamped as a severity [InlineData("Severity", "0", 1)] // clamp to OPC UA min 1 [InlineData("Severity", "5000", 1000)] // clamp to OPC UA max 1000 [InlineData(null, null, 1)] // missing → default min severity @@ -54,6 +54,95 @@ public sealed class EventMapperTests Assert.Equal((ushort)expected, EventMapper.ToHistoricalEvent(e).Severity); } + // ---- Gitea #886: AVEVA-shaped rows (lower-case keys) ------------------------------------------------------------- + + /// An AVEVA A&E alarm row as vd03's HistorianGateway returns it (2026-11-12 probe, trimmed). + private static HistorianEvent AvevaAlarmRow(string priority = "500") + { + var e = new HistorianEvent + { + Id = "001c0040-0025-0004-5b00-74003f005400", Type = "Alarm.Set", SourceName = "Oven_007.Temp", + Namespace = "Application Server", EventTime = Ts(2026, 11, 5, 15, 19, 52), ReceivedTime = Ts(2026, 11, 5, 15, 19, 53), + }; + e.Properties["alarm_id"] = "001c0040-0025-0004-5b00-74003f005400"; + e.Properties["alarm_state"] = "UNACK_ALM"; + e.Properties["comment"] = "Oven 7 temperature"; + e.Properties["priority"] = priority; + e.Properties["severity"] = "2"; // AVEVA's own 1..4 scale: never the OPC UA Severity + e.Properties["source_conditionvariable"] = "Oven_007.TempHiAlarm"; + e.Properties["source_object"] = "Oven_007"; + e.Properties["source_processvariable"] = "Oven_007.Temp"; + return e; + } + + [Fact] + public void An_aveva_alarm_row_maps_priority_alarm_id_and_its_condition_reference() + { + var h = EventMapper.ToHistoricalEvent(AvevaAlarmRow()); + + Assert.Equal((ushort)750, h.Severity); // priority 500 → the live Galaxy table's High bucket, not AVEVA's severity "2" + Assert.Equal("001c0040-0025-0004-5b00-74003f005400", h.AlarmId); + Assert.Equal("Oven_007.TempHiAlarm", h.ConditionReference); + Assert.Equal("TempHiAlarm", h.ConditionName); // the AVEVA alarm's name until the server resolves a live condition + Assert.Equal("Alarm.Set", h.Message); // no message property: the event type, as before + Assert.Equal("Oven_007.Temp", h.SourceName); + } + + [Theory] + [InlineData("1", 250)] + [InlineData("249", 250)] + [InlineData("250", 500)] + [InlineData("499", 500)] + [InlineData("500", 750)] + [InlineData("749", 750)] + [InlineData("750", 1000)] + [InlineData("999", 1000)] // a System.Stop/Start row (priority 999, AVEVA severity 0) — lmxopcua#887 tracks the direction + public void An_aveva_priority_maps_through_the_live_galaxy_table(string priority, int expected) + { + var h = EventMapper.ToHistoricalEvent(AvevaAlarmRow(priority)); + Assert.Equal((ushort)expected, h.Severity); + // The live Galaxy path maps the same raw value through the same table, so history and live agree. + Assert.Equal(expected, ZB.MOM.WW.OtOpcUa.Core.Abstractions.GalaxyAlarmPriority.Map(int.Parse(priority)).OpcUaSeverity); + } + + [Fact] + public void OtOpcUas_own_event_read_back_from_aveva_with_lower_case_keys_keeps_its_fields() + { + // AVEVA lower-cases the keys OtOpcUa writes (AlarmName → alarmname, …): seen on vd03 2026-11-12. + var e = new HistorianEvent { Id = "E", Type = "LimitAlarm", SourceName = "Area/Line/Eq", + EventTime = Ts(2026, 6, 27, 3, 43, 44), ReceivedTime = Ts(2026, 6, 27, 3, 43, 44) }; + e.Properties["alarmid"] = "Plant/Modbus/dev1/temp_hi"; + e.Properties["alarmname"] = "HiTemp"; + e.Properties["eventkind"] = "Activated"; + e.Properties["message"] = "Temperature high"; + e.Properties["severity"] = "700"; + + var h = EventMapper.ToHistoricalEvent(e); + + Assert.Equal("Plant/Modbus/dev1/temp_hi", h.AlarmId); + Assert.Equal("HiTemp", h.ConditionName); + Assert.Equal("Temperature high", h.Message); + Assert.Equal((ushort)700, h.Severity); + Assert.Null(h.ConditionReference); + } + + [Fact] + public void An_aveva_system_event_has_no_condition() + { + var e = new HistorianEvent { Id = "S", Type = "System.Stop", SourceName = "Oven_007", + EventTime = Ts(2026, 11, 10, 11, 32, 35), ReceivedTime = Ts(2026, 11, 10, 11, 32, 35) }; + e.Properties["priority"] = "999"; + e.Properties["severity"] = "0"; + e.Properties["source_object"] = "Oven_007"; + + var h = EventMapper.ToHistoricalEvent(e); + + Assert.Null(h.ConditionName); + Assert.Null(h.ConditionReference); + Assert.Null(h.AlarmId); + Assert.Equal("System.Stop", h.Message); + } + // Ts(...) builds a Google.Protobuf.WellKnownTypes.Timestamp from UTC parts. private static Timestamp Ts(int y, int mo, int d, int h, int mi, int s) => Timestamp.FromDateTime(new DateTime(y, mo, d, h, mi, s, DateTimeKind.Utc)); diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/AvevaGatewayHistoryLiveGateTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/AvevaGatewayHistoryLiveGateTests.cs index 27c9ad8d4..8c6f84663 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/AvevaGatewayHistoryLiveGateTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/AvevaGatewayHistoryLiveGateTests.cs @@ -33,7 +33,9 @@ namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests; /// OTOPCUA_AVEVA_GATE_OUT (output directory for aveva-gate-*.json), OTOPCUA_AVEVA_GATE_LABEL /// (a run label recorded in the output, e.g. the SDK version), OTOPCUA_AVEVA_GATE_TAGS (comma list of /// historized analog tags), OTOPCUA_AVEVA_GATE_WINDOWS (start/end;start/end, ISO-8601 UTC), -/// OTOPCUA_AVEVA_GATE_EVENT_SOURCES (comma list) and OTOPCUA_AVEVA_GATE_EVENT_WINDOW. The defaults are the DEV +/// OTOPCUA_AVEVA_GATE_EVENT_SOURCES (comma list), OTOPCUA_AVEVA_GATE_EVENT_WINDOW and +/// OTOPCUA_AVEVA_GATE_GALAXY_REFS (#886: comma list of Galaxy alarm references whose native conditions are wired to +/// one equipment folder that is not named after any of their objects). The defaults are the DEV /// (wonder-app-vd03 / wonder-sql-vd03) tags and fixed past windows used for the 2026-11-12 run. /// The local deterministic fixtures are NOT this test; only a real gateway in front of a real AVEVA /// Historian makes its output AVEVA evidence. @@ -53,6 +55,12 @@ public sealed class AvevaGatewayHistoryLiveGateTests // it stays GoodNoData. Recorded (wire must equal the backend), not required to be non-empty. private const string DefaultEventSources = "DevWinPlatform,DevWinPlatform.CPULoadAvg"; private const string DefaultEventWindow = "2026-11-05T15:00:00Z/2026-11-12T00:00:00Z"; + + // #886: a REAL equipment folder — its id is not a Galaxy object name — whose native conditions dial these Galaxy alarm + // references (attributeRef). Two objects with A&E history on DEV in the event window: the platform's CPU-load alarm + // and a ZIPSEN Left_001 machine alarm. The folder must serve both objects' events and attribute alarms. + private const string DefaultGalaxyAlarmRefs = "DevWinPlatform.CPULoadHiAlarm,Left_001.FRR_Warning"; + private const string GalaxyEquipment = "EQ-886-ZipsenLeft"; private const string MissingTag = "OtOpcUa.Gate801.DoesNotExist"; private const double IntervalMs = 60_000; @@ -99,6 +107,8 @@ public sealed class AvevaGatewayHistoryLiveGateTests .Split(',', StringSplitOptions.RemoveEmptyEntries | StringSplitOptions.TrimEntries); var eventWindow = ParseWindows(Env("OTOPCUA_AVEVA_GATE_EVENT_WINDOW") ?? DefaultEventWindow)[0]; var sandboxTag = Env("HISTGW_WRITE_SANDBOX_TAG"); + var galaxyRefs = (Env("OTOPCUA_AVEVA_GATE_GALAXY_REFS") ?? DefaultGalaxyAlarmRefs) + .Split(',', StringSplitOptions.RemoveEmptyEntries | StringSplitOptions.TrimEntries); // The production backend path: the same registry + provider + factory the Host uses for Backend=Gateway. var options = new ServerHistorianOptions @@ -142,6 +152,23 @@ public sealed class AvevaGatewayHistoryLiveGateTests AddressSpaceRealm.Raw, isNative: true); } + // #886: the applier's shape for a Galaxy deployment — each native condition under its raw device folder, bound to + // its Galaxy alarm reference and wired to the one equipment folder that references both. + rig.Sink.EnsureFolder(GalaxyEquipment, parentNodeId: null, displayName: "ZIPSEN Left line", realm: AddressSpaceRealm.Uns); + var galaxyConditions = new Dictionary(StringComparer.OrdinalIgnoreCase); + foreach (var reference in galaxyRefs) + { + var objectName = GalaxyAttributeRef.ObjectNameOf(reference)!; + var name = reference[(reference.IndexOf('.') + 1)..].Replace(".", "_", StringComparison.Ordinal); + var deviceFolder = $"Plant/Galaxy/{objectName}"; + var rawPath = $"{deviceFolder}/{name}"; + rig.Sink.EnsureFolder(deviceFolder, parentNodeId: null, displayName: objectName, realm: AddressSpaceRealm.Raw); + rig.Sink.MaterialiseAlarmCondition(rawPath, deviceFolder, name, "OffNormalAlarm", 500, AddressSpaceRealm.Raw, isNative: true); + rig.Sink.BindNativeAlarmHistory(rawPath, AddressSpaceRealm.Raw, reference); + rig.Sink.WireAlarmNotifiers(rawPath, AddressSpaceRealm.Raw, [GalaxyEquipment], AddressSpaceRealm.Uns); + galaxyConditions[reference] = (rawPath, name); + } + using var session = await rig.ConnectAsync(anonymous: true, ct); var ns = (ushort)session.NamespaceUris.GetIndex(V3NodeIds.RawNamespaceUri); NodeId Node(string id) => new(id, ns); @@ -299,6 +326,78 @@ public sealed class AvevaGatewayHistoryLiveGateTests } } + // ---- #886: a real equipment folder serves the AVEVA history of every Galaxy object behind its native conditions. + { + var (start, end) = eventWindow; + var w = $"{start:o}/{end:o}"; + var objects = galaxyRefs.Select(r => GalaxyAttributeRef.ObjectNameOf(r)!).Distinct(StringComparer.Ordinal).ToList(); + var discovered = await source.ReadEventsAsync(null, start, end, 0, ct); + bool OfAnObject(string? s) => s is not null && objects.Any(o => s == o || s.StartsWith(o + ".", StringComparison.Ordinal)); + var expected = discovered.Events.Where(e => OfAnObject(e.SourceName)).ToList(); + // The backend's own mapping of each event (per object, as the folder reads them): severity and AVEVA's condition. + var direct = new Dictionary(StringComparer.Ordinal); + foreach (var o in objects) + foreach (var e in (await source.ReadEventsAsync(o, start, end, 0, ct)).Events) direct.TryAdd(e.EventId, e); + var folderOwn = await source.ReadEventsAsync(GalaxyEquipment, start, end, 0, ct); + + var uns = new NodeId(GalaxyEquipment, (ushort)session.NamespaceUris.GetIndex(V3NodeIds.UnsNamespaceUri)); + var all = await ReadGalaxyEventsAsync(session, uns, start, end, 0, ct); + var paged1 = await ReadGalaxyEventsAsync(session, uns, start, end, 1, ct); + var paged7 = await ReadGalaxyEventsAsync(session, uns, start, end, 7, ct); + var reverse = await ReadGalaxyEventsAsync(session, uns, end, start, 0, ct); + var reverse5 = await ReadGalaxyEventsAsync(session, uns, end, start, 5, ct); + + var bySource = string.Join(",", all.Rows.GroupBy(r => r.Event.Source ?? "").OrderBy(g => g.Key, StringComparer.Ordinal) + .Select(g => $"{g.Key}={g.Count()}")); + var resolved = all.Rows.Count(r => r.SourceNode is not null); + report.Add(new Case("events886.folderOwnSourceOnly", GalaxyEquipment, w, "direct", "Good", + $"events={folderOwn.Events.Count} (what the folder read before #886)", null)); + report.Add(new Case("events886.all", GalaxyEquipment, w, "wire", all.Status, + $"pages={all.Pages} objects={string.Join('+', objects)} bySource={bySource} sourceNodeResolved={resolved}", null, null, + all.Rows.Select(r => r.Event).ToList())); + report.Add(new Case("events886.paged1", GalaxyEquipment, w, "wire", paged1.Status, $"pages={paged1.Pages}", null, null, + paged1.Rows.Select(r => r.Event).ToList())); + report.Add(new Case("events886.paged7", GalaxyEquipment, w, "wire", paged7.Status, $"pages={paged7.Pages}", null, null, + paged7.Rows.Select(r => r.Event).ToList())); + report.Add(new Case("events886.reverse", GalaxyEquipment, w, "wire", reverse.Status, $"pages={reverse.Pages}", null, null, + reverse.Rows.Select(r => r.Event).ToList())); + report.Add(new Case("events886.reverse5", GalaxyEquipment, w, "wire", reverse5.Status, $"pages={reverse5.Pages}", null, null, + reverse5.Rows.Select(r => r.Event).ToList())); + report.Add(new Case("events886.conditions", GalaxyEquipment, w, "wire", all.Status, + string.Join(";", all.Rows.Where(r => r.SourceNode is not null).GroupBy(r => (r.SourceNode, r.ConditionName)) + .Select(g => $"{g.Key.SourceNode}|{g.Key.ConditionName}={g.Count()}")), null)); + + foreach (var o in objects) + Check(failures, expected.Any(e => e.SourceName == o || e.SourceName!.StartsWith(o + ".", StringComparison.Ordinal)), + $"events886 {o} {w}: expected stored events for every object behind the folder"); + Check(failures, folderOwn.Events.Count == 0, $"events886: the folder id {GalaxyEquipment} must name no AVEVA source"); + var expectedIds = expected.Select(e => e.EventId).OrderBy(id => id, StringComparer.Ordinal).ToList(); + Check(failures, all.Rows.Select(r => r.Event.Id).OrderBy(id => id, StringComparer.Ordinal).SequenceEqual(expectedIds), + $"events886 {w}: wire {all.Rows.Count} events vs {expectedIds.Count} of {string.Join('/', objects)} or their attributes"); + Check(failures, all.Rows.Zip(all.Rows.Skip(1)).All(p => p.First.Event.Time <= p.Second.Event.Time), + "events886: a forward read is in time order across objects"); + var forwardIds = all.Rows.Select(r => r.Event.Id).ToList(); + Check(failures, paged1.Rows.Select(r => r.Event.Id).SequenceEqual(forwardIds), "events886: 1-event pages equal the whole read"); + Check(failures, paged7.Rows.Select(r => r.Event.Id).SequenceEqual(forwardIds), "events886: 7-event pages equal the whole read"); + var reverseIds = expected.OrderByDescending(e => e.EventTimeUtc).ThenByDescending(e => e.EventId, StringComparer.Ordinal) + .Select(e => e.EventId).ToList(); + Check(failures, reverse.Rows.Select(r => r.Event.Id).SequenceEqual(reverseIds), + "events886: a reverse read is newest first (ties by EventId, descending)"); + Check(failures, reverse5.Rows.Select(r => r.Event.Id).SequenceEqual(reverseIds), "events886: 5-event reverse pages equal the reverse read"); + foreach (var r in all.Rows) + { + if (!direct.TryGetValue(r.Event.Id!, out var d)) { failures.Add($"events886: {r.Event.Id} not in the per-object read"); continue; } + Check(failures, r.Event.Severity == d.Severity, $"events886: {r.Event.Id} severity {r.Event.Severity} vs backend {d.Severity}"); + var live = d.ConditionReference is { } cr && galaxyConditions.TryGetValue(cr, out var c) ? c : ((string RawPath, string Name)?)null; + Check(failures, r.SourceNode == live?.RawPath, + $"events886: {r.Event.Id} ({d.ConditionReference}) SourceNode {r.SourceNode ?? "null"} vs {live?.RawPath ?? "null"}"); + if (live is { } lc) + Check(failures, r.ConditionName == lc.Name, $"events886: {r.Event.Id} ConditionName {r.ConditionName} vs {lc.Name}"); + } + + Check(failures, resolved > 0, "events886: at least one row resolves to its live condition"); + } + // ---- HistoryUpdate: refused by the server (no updater, no HistoryWrite bit). Sandbox-bound node only. if (sandboxNode is null) { @@ -463,6 +562,49 @@ public sealed class AvevaGatewayHistoryLiveGateTests return new EventRead(status, rows, pages); } + /// #886: an events read that also returns the SourceNode and ConditionName (paged by continuation points). + private static async Task ReadGalaxyEventsAsync(ISession session, NodeId node, DateTime start, DateTime end, uint page, + CancellationToken ct) + { + var filter = new EventFilter(); + foreach (var field in new[] { "EventId", "Time", "ReceiveTime", "SourceName", "Message", "Severity", "SourceNode", "ConditionName" }) + { + var type = field == "ConditionName" ? ObjectTypeIds.ConditionType : ObjectTypeIds.BaseEventType; + filter.SelectClauses += new SimpleAttributeOperand(type, new QualifiedName(field)) { AttributeId = Attributes.Value }; + } + + var details = new ReadEventDetails { StartTime = start, EndTime = end, NumValuesPerNode = page, Filter = filter }; + var rows = new List(); + var pages = 0; + ByteString cp = default; + string status; + do + { + var response = await session.HistoryReadAsync(null, new ExtensionObject(details), TimestampsToReturn.Both, false, + new HistoryReadValueId[] { new() { NodeId = node, ContinuationPoint = cp } }, ct); + var result = response.Results.ToList().ShouldHaveSingleItem(); + status = Name(result.StatusCode); + cp = result.ContinuationPoint; + pages++; + if (!result.HistoryData.IsNull && ExtensionObject.ToEncodeable(result.HistoryData) is HistoryEvent he) + foreach (var e in he.Events.ToList()) + { + var f = e.EventFields.ToList(); + rows.Add(new GalaxyEventRow( + new EventRow( + f[0].ClrValue() is byte[] id ? System.Text.Encoding.UTF8.GetString(id) : Fmt(f[0].ClrValue()), + f[1].ClrValue() as DateTime? ?? default, f[2].ClrValue() as DateTime? ?? default, + f[3].ClrValue() as string, f[4].ClrValue() is LocalizedText lt ? lt.Text : Fmt(f[4].ClrValue()), + Convert.ToUInt16(f[5].ClrValue() ?? 0, CultureInfo.InvariantCulture)), + f[6].ClrValue() is NodeId sn && !sn.IsNull ? sn.IdentifierAsString : null, + f[7].ClrValue() as string)); + } + if (status.StartsWith("Bad", StringComparison.Ordinal)) break; + } + while (!cp.IsEmpty && pages < 10_000); + return new GalaxyEventRead(status, rows, pages); + } + private static async Task> ReadAttributesAsync(ISession session, NodeId node, CancellationToken ct) { var ids = new (string Name, uint Id)[] @@ -586,6 +728,10 @@ public sealed class AvevaGatewayHistoryLiveGateTests private sealed record EventRead(string Status, List Rows, int Pages); + private sealed record GalaxyEventRow(EventRow Event, string? SourceNode, string? ConditionName); + + private sealed record GalaxyEventRead(string Status, List Rows, int Pages); + private sealed record Case( string Name, string? Tag, string? Window, string Path, string Status, string? Note, Dictionary? Attributes, List? Rows = null, List? Events = null); diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/GalaxyEquipmentEventHistoryWireTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/GalaxyEquipmentEventHistoryWireTests.cs new file mode 100644 index 000000000..410efef9a --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/GalaxyEquipmentEventHistoryWireTests.cs @@ -0,0 +1,322 @@ +using System.Runtime.CompilerServices; +using Google.Protobuf.WellKnownTypes; +using Microsoft.Extensions.Logging.Abstractions; +using Opc.Ua; +using Opc.Ua.Client; +using Shouldly; +using Xunit; +using ZB.MOM.WW.HistorianGateway.Contracts.Grpc; +using ZB.MOM.WW.OtOpcUa.Commons.OpcUa; +using ZB.MOM.WW.OtOpcUa.Configuration.Enums; +using ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway; +using static ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests.ServerEventRig; + +namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests; + +/// +/// Gitea #886 over a real SDK session, with the production Gateway backend on a deterministic AVEVA-shaped gateway: a +/// REAL equipment folder — whose id (EqA) is not a Galaxy object name — serves the AVEVA history of the Galaxy +/// objects behind its native conditions, here two (Oven_007 and OvenDoor_07): each object's own events +/// and its attribute alarms, never a prefix sibling (Oven_0070), merged in time order across paging and reverse +/// reads, with the live condition's SourceNode / ConditionName and the live Galaxy severity, and only behind the +/// folder's own HistoryRead verdict. +/// +/// +/// Topology, as the applier builds it from a Galaxy deployment: raw folder Plant/Galaxy/Oven7 holds two native +/// conditions, TempHi (attributeRef Oven_007.TempHiAlarm) and DoorOpen +/// (OvenDoor_07.DoorAlarm), both wired to equipment EqA; raw folder Plant/Galaxy/Line3 holds +/// Jam (Line3_Conveyor.JamAlarm), wired to EqB. The fake gateway filters a per-source read more +/// loosely than the real one (Source_Object begins with the source), so the sibling's rows reach OtOpcUa and +/// their absence is OtOpcUa's filter at work. +/// +public sealed class GalaxyEquipmentEventHistoryWireTests +{ + private const string OvenFolder = "Plant/Galaxy/Oven7"; + private const string LineFolder = "Plant/Galaxy/Line3"; + private const string TempHi = OvenFolder + "/TempHi"; + private const string DoorOpen = OvenFolder + "/DoorOpen"; + private const string Jam = LineFolder + "/Jam"; + + private static readonly DateTime Epoch = new(2026, 11, 10, 11, 0, 0, DateTimeKind.Utc); + + /// The stored AVEVA rows. A null condition variable is an object-level system event. + private static readonly AvevaRow[] Stored = + [ + new("o-stop", "Oven_007", "Oven_007", "System.Stop", 0, 999, null), + new("l-set", "Line3_Conveyor.Belt", "Line3_Conveyor", "Alarm.Set", 1, 300, "Line3_Conveyor.JamAlarm"), + new("o-temp-set", "Oven_007.Temp", "Oven_007", "Alarm.Set", 2, 500, "Oven_007.TempHiAlarm"), + new("d-set", "OvenDoor_07.Door", "OvenDoor_07", "Alarm.Set", 2, 700, "OvenDoor_07.DoorAlarm"), // a tie across objects + new("a-oven-tie", "Oven_007.Temp", "Oven_007", "Alarm.Ack", 2, 500, "Oven_007.TempHiAlarm"), // and within one, stored later + new("o-press-set", "Oven_007.Pressure", "Oven_007", "Alarm.Set", 3, 100, "Oven_007.PressureHiAlarm"), // no live condition here + new("s-set", "Oven_0070.Temp", "Oven_0070", "Alarm.Set", 4, 500, "Oven_0070.TempHiAlarm"), // prefix sibling + new("d-clear", "OvenDoor_07.Door", "OvenDoor_07", "Alarm.Clear", 5, 700, "OvenDoor_07.DoorAlarm"), + new("o-temp-clear", "Oven_007.Temp", "Oven_007", "Alarm.Clear", 6, 500, "oven_007.temphialarm"), // Galaxy names: any case + new("l-clear", "Line3_Conveyor.Belt", "Line3_Conveyor", "Alarm.Clear", 7, 300, "Line3_Conveyor.JamAlarm"), + new("o-start", "Oven_007", "Oven_007", "System.Start", 8, 999, null), + ]; + + /// EqA's events forward: by time, then source (the folder's ordinal-sorted list: OvenDoor_07 before + /// Oven_007), then the order the gateway returned them in — so a-oven-tie follows o-temp-set. + private static readonly string[] EqAEvents = + ["o-stop", "d-set", "o-temp-set", "a-oven-tie", "o-press-set", "d-clear", "o-temp-clear", "o-start"]; + + /// EqA's events reverse: newest first; the Gateway's walked-back reverse read orders ties by EventId, + /// descending, as it did before #886. + private static readonly string[] EqAEventsReverse = + ["o-start", "o-temp-clear", "d-clear", "o-press-set", "o-temp-set", "d-set", "a-oven-tie", "o-stop"]; + + [Fact] + public async Task An_equipment_folder_serves_the_aveva_history_of_every_galaxy_object_behind_it() + { + var ct = TestContext.Current.CancellationToken; + await using var rig = await StartAsync("GalaxyEqHistory", encrypted: false, ct); + var gateway = new AvevaShapedGateway(); + rig.Nm.HistorianDataSource = new GatewayHistorianDataSource(gateway, NullLogger.Instance); + Materialise(rig); + rig.Enforce(false, [Key()]); + + using var session = await rig.ConnectAsync(anonymous: true, ct); + var all = await ReadAsync(session, Uns(session, EqA), forward: true, page: 0, ct); + + all.Status.ShouldBe(StatusCodes.Good); + all.Rows.Select(r => r.Id).ShouldBe(EqAEvents); + gateway.Requested.ShouldBe(new[] { EqA, "OvenDoor_07", "Oven_007" }, + "the folder's own source and one request per Galaxy object behind it (no multi-source request exists)"); + gateway.Yielded.ShouldContain("s-set", "the sibling's rows reached OtOpcUa, so their absence is its filter's doing"); + + var byId = all.Rows.ToDictionary(r => r.Id); + // Each event keeps the source the historian reported. + byId["o-temp-set"].Source.ShouldBe("Oven_007.Temp"); + byId["o-stop"].Source.ShouldBe("Oven_007"); + // A row naming a condition that fires on this folder carries that live condition (any-case reference). + byId["o-temp-set"].SourceNode.ShouldBe(TempHi); + byId["o-temp-clear"].SourceNode.ShouldBe(TempHi); + byId["o-temp-set"].ConditionName.ShouldBe("TempHi"); + byId["d-set"].SourceNode.ShouldBe(DoorOpen); + byId["d-set"].ConditionName.ShouldBe("DoorOpen"); + // An alarm of the object with no condition wired here: served (object rule), but names no node. + byId["o-press-set"].SourceNode.ShouldBeNull(); + byId["o-press-set"].ConditionName.ShouldBe("PressureHiAlarm"); + // A system event has no condition at all. + byId["o-stop"].SourceNode.ShouldBeNull(); + byId["o-stop"].ConditionName.ShouldBeNull(); + // Severity follows the live Galaxy table (priority 500 → 750, 700 → 750, 100 → 250, 999 → 1000), not Severity 1. + byId["o-temp-set"].Severity.ShouldBe((ushort)750); + byId["d-set"].Severity.ShouldBe((ushort)750); + byId["o-press-set"].Severity.ShouldBe((ushort)250); + byId["o-stop"].Severity.ShouldBe((ushort)1000); + + // The raw device folder the conditions hang under reads the same objects (live: a condition fires on its parent). + var raw = await ReadAsync(session, Raw(session, OvenFolder), forward: true, page: 0, ct); + raw.Rows.Select(r => r.Id).ShouldBe(EqAEvents); + } + + [Fact] + public async Task Paging_and_reverse_reads_walk_the_merged_sources_exactly_once() + { + var ct = TestContext.Current.CancellationToken; + await using var rig = await StartAsync("GalaxyEqHistoryPaging", encrypted: false, ct); + var gateway = new AvevaShapedGateway(); + rig.Nm.HistorianDataSource = new GatewayHistorianDataSource(gateway, NullLogger.Instance); + Materialise(rig); + rig.Enforce(false, [Key()]); + + using var session = await rig.ConnectAsync(anonymous: true, ct); + var node = Uns(session, EqA); + foreach (var page in new uint[] { 1, 2, 3, 4 }) + { + var paged = await ReadAsync(session, node, forward: true, page, ct); + paged.Rows.Select(r => r.Id).ShouldBe(EqAEvents, $"forward, {page} a page"); + paged.Pages.ShouldBeGreaterThan(1); + + var reverse = await ReadAsync(session, node, forward: false, page, ct); + reverse.Rows.Select(r => r.Id).ShouldBe(EqAEventsReverse, $"reverse, {page} a page (newest first, ties descending)"); + } + + var reverseWhole = await ReadAsync(session, node, forward: false, page: 0, ct); + reverseWhole.Rows.Select(r => r.Id).ShouldBe(EqAEventsReverse); + } + + [Fact] + public async Task With_the_acl_on_a_denied_folder_reads_none_of_its_galaxy_objects() + { + var ct = TestContext.Current.CancellationToken; + await using var rig = await StartAsync("GalaxyEqHistoryAcl", encrypted: false, ct); + var gateway = new AvevaShapedGateway(); + rig.Nm.HistorianDataSource = new GatewayHistorianDataSource(gateway, NullLogger.Instance); + Materialise(rig); + // Anonymous is granted EqB only: EqA, behind which sit Oven_007 and OvenDoor_07, is denied. + rig.Enforce(true, [Key()], EquipmentGrant(EqB, NodePermissions.ReadOnly)); + + using var session = await rig.ConnectAsync(anonymous: true, ct); + var denied = await ReadAsync(session, Uns(session, EqA), forward: true, page: 0, ct); + + denied.Status.ShouldBe(StatusCodes.BadUserAccessDenied); + denied.HistoryDataIsNull.ShouldBeTrue("a denied folder discloses none of its events"); + gateway.Requested.ShouldBeEmpty("the folder verdict is decided before any source is read (fail closed)"); + + // Positive control: the same session, on the folder it IS granted, gets that folder's Galaxy object. + var granted = await ReadAsync(session, Uns(session, EqB), forward: true, page: 0, ct); + granted.Status.ShouldBe(StatusCodes.Good); + granted.Rows.Select(r => (r.Id, r.SourceNode)).ShouldBe(new (string, string?)[] { ("l-set", Jam), ("l-clear", Jam) }); + gateway.Requested.ShouldBe(new[] { EqB, "Line3_Conveyor" }); + } + + [Fact] + public async Task A_rebuild_forgets_the_galaxy_sources_until_the_conditions_are_bound_again() + { + var ct = TestContext.Current.CancellationToken; + await using var rig = await StartAsync("GalaxyEqHistoryRebuild", encrypted: false, ct); + var gateway = new AvevaShapedGateway(); + rig.Nm.HistorianDataSource = new GatewayHistorianDataSource(gateway, NullLogger.Instance); + Materialise(rig); + rig.Enforce(false, [Key()]); + + // The redeploy re-creates the same conditions and wiring, but (here) binds no Galaxy reference: the folder is back + // to its own source alone — a stale binding would keep reading the old objects. + rig.Sink.RebuildAddressSpace(); + Materialise(rig, bind: false); + + using var session = await rig.ConnectAsync(anonymous: true, ct); + var read = await ReadAsync(session, Uns(session, EqA), forward: true, page: 0, ct); + + read.Status.ShouldBe(StatusCodes.GoodNoData); + gateway.Requested.ShouldBe(new[] { EqA }); + } + + /// Materialises the topology the applier builds from the Galaxy deployment (see the class remarks). + private static void Materialise(ServerEventRig rig, bool bind = true) + { + rig.Sink.EnsureFolder(OvenFolder, parentNodeId: null, displayName: "Oven7", realm: AddressSpaceRealm.Raw); + rig.Sink.EnsureFolder(LineFolder, parentNodeId: null, displayName: "Line3", realm: AddressSpaceRealm.Raw); + rig.EnsureEquipment(); + foreach (var (id, folder, name, reference, equipment) in new[] + { + (TempHi, OvenFolder, "TempHi", "Oven_007.TempHiAlarm", EqA), + (DoorOpen, OvenFolder, "DoorOpen", "OvenDoor_07.DoorAlarm", EqA), + (Jam, LineFolder, "Jam", "Line3_Conveyor.JamAlarm", EqB), + }) + { + rig.Sink.MaterialiseAlarmCondition(id, folder, name, "OffNormalAlarm", 700, AddressSpaceRealm.Raw, isNative: true); + if (bind) rig.Sink.BindNativeAlarmHistory(id, AddressSpaceRealm.Raw, reference); + rig.Sink.WireAlarmNotifiers(id, AddressSpaceRealm.Raw, [equipment], AddressSpaceRealm.Uns); + } + } + + private static NodeId Uns(ISession session, string id) => + new(id, (ushort)session.NamespaceUris.GetIndex(V3NodeIds.UnsNamespaceUri)); + + private static NodeId Raw(ISession session, string id) => + new(id, (ushort)session.NamespaceUris.GetIndex(V3NodeIds.RawNamespaceUri)); + + private static async Task ReadAsync(ISession session, NodeId node, bool forward, uint page, CancellationToken ct) + { + var filter = new EventFilter(); + foreach (var leaf in new[] { "EventId", "SourceName", "SourceNode", "Severity", "ConditionName" }) + { + var type = leaf == "ConditionName" ? ObjectTypeIds.ConditionType : ObjectTypeIds.BaseEventType; + filter.SelectClauses += new SimpleAttributeOperand(type, new QualifiedName(leaf)) { AttributeId = Attributes.Value }; + } + + var (from, to) = (Epoch.AddMinutes(-1), Epoch.AddMinutes(1)); + var details = new ReadEventDetails + { + StartTime = forward ? from : to, EndTime = forward ? to : from, NumValuesPerNode = page, Filter = filter, + }; + var rows = new List(); + ByteString point = default; + var pages = 0; + StatusCode first = StatusCodes.Good; + var historyDataIsNull = true; + do + { + var response = await session.HistoryReadAsync(null, new ExtensionObject(details), TimestampsToReturn.Both, false, + new HistoryReadValueId[] { new() { NodeId = node, ContinuationPoint = point } }, ct); + var result = response.Results.ToList().ShouldHaveSingleItem(); + if (pages++ == 0) { first = result.StatusCode; historyDataIsNull = result.HistoryData.IsNull; } + if (StatusCode.IsBad(result.StatusCode)) break; + if (!result.HistoryData.IsNull && ExtensionObject.ToEncodeable(result.HistoryData) is HistoryEvent he) + foreach (var e in he.Events.ToList()) + { + var f = e.EventFields.ToList(); + rows.Add(new Row( + System.Text.Encoding.UTF8.GetString(f[0].GetByteString().Span), + f[1].GetString(), + f[2].IsNull ? null : f[2].GetNodeId().IdentifierAsString, + f[3].GetUInt16(), + f[4].IsNull ? null : f[4].GetString())); + } + + point = result.ContinuationPoint; + } while (!point.IsEmpty && pages < 60); + + return new EventRead(first, historyDataIsNull, rows, pages); + } + + private sealed record AvevaRow(string Id, string Source, string Object, string Type, int Second, int Priority, string? ConditionVariable); + + private sealed record Row(string Id, string? Source, string? SourceNode, ushort Severity, string? ConditionName); + + private sealed record EventRead(StatusCode Status, bool HistoryDataIsNull, List Rows, int Pages); + + /// + /// A deterministic gateway answering ReadEvents with AVEVA-shaped rows — SourceName = + /// Source_ProcessVariable and lower-case properties as vd03's HistorianGateway returns them — filtered loosely + /// on source_object, in event-time order with no intra-timestamp order guarantee (here: storage order). + /// + private sealed class AvevaShapedGateway : IHistorianGatewayClient + { + private readonly List _requested = []; + private readonly List _yielded = []; + + public IReadOnlyList Requested { get { lock (_requested) return _requested.ToList(); } } + + public IReadOnlyList Yielded { get { lock (_yielded) return _yielded.ToList(); } } + + public async IAsyncEnumerable ReadEventsAsync( + string? sourceName, DateTime startUtc, DateTime endUtc, int maxEvents, [EnumeratorCancellation] CancellationToken ct) + { + lock (_requested) _requested.Add(sourceName); + await Task.Yield(); + foreach (var s in Stored.OrderBy(s => s.Second)) + { + var at = Epoch.AddSeconds(s.Second); + if (at < startUtc || at > endUtc) continue; // the gateway's end is inclusive + if (sourceName is not null && !s.Object.StartsWith(sourceName, StringComparison.Ordinal)) continue; + lock (_yielded) _yielded.Add(s.Id); + var ts = Timestamp.FromDateTime(at); + var e = new HistorianEvent + { + Id = s.Id, SourceName = s.Source, Type = s.Type, Namespace = "Application Server", EventTime = ts, ReceivedTime = ts, + }; + e.Properties["source_object"] = s.Object; + e.Properties["source_processvariable"] = s.Source; + e.Properties["priority"] = s.Priority.ToString(System.Globalization.CultureInfo.InvariantCulture); + e.Properties["severity"] = s.ConditionVariable is null ? "0" : "2"; // AVEVA's own scale: never used + if (s.ConditionVariable is not null) + { + e.Properties["source_conditionvariable"] = s.ConditionVariable; + e.Properties["alarm_id"] = $"guid-{s.ConditionVariable.ToLowerInvariant()}"; + } + + yield return e; + } + } + + public Task TagExistsAsync(string tag, CancellationToken ct) => throw new NotSupportedException(); + public IAsyncEnumerable ReadRawAsync(string tag, DateTime startUtc, DateTime endUtc, int maxValues, CancellationToken ct) => + throw new NotSupportedException(); + public IAsyncEnumerable ReadAggregateAsync(string tag, DateTime startUtc, DateTime endUtc, + RetrievalMode mode, TimeSpan interval, CancellationToken ct) => throw new NotSupportedException(); + public Task> ReadAtTimeAsync(string tag, IReadOnlyList timestampsUtc, CancellationToken ct) => + throw new NotSupportedException(); + public Task WriteLiveValuesAsync(string tag, IReadOnlyList values, CancellationToken ct) => + throw new NotSupportedException(); + public Task SendEventAsync(HistorianEvent evt, CancellationToken ct) => throw new NotSupportedException(); + public Task EnsureTagsAsync(IReadOnlyList definitions, CancellationToken ct) => + throw new NotSupportedException(); + public Task ProbeAsync(CancellationToken ct) => Task.FromResult(true); + public Task GetConnectionStatusAsync(CancellationToken ct) => Task.FromResult(new ConnectionStatus()); + public ValueTask DisposeAsync() => ValueTask.CompletedTask; + } +} diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/AddressSpaceApplierRawUnsTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/AddressSpaceApplierRawUnsTests.cs index 036a642d2..e83c14f0b 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/AddressSpaceApplierRawUnsTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/AddressSpaceApplierRawUnsTests.cs @@ -154,6 +154,38 @@ public sealed class AddressSpaceApplierRawUnsTests w.FolderNodeIds.ShouldBe(new[] { "EQ-1", "EQ-2" }, ignoreOrder: true); } + /// Gitea #886: a native Galaxy alarm's AVEVA reference reaches the sink, bound to its condition after the + /// condition is materialised and before its notifiers are wired; a native alarm with no reference binds nothing. + [Fact] + public void MaterialiseRawSubtree_binds_a_galaxy_alarms_aveva_reference_to_its_condition() + { + var sink = new RealmRecordingSink(); + var composition = new AddressSpaceComposition( + new[] { new UnsAreaProjection("a1", "filling") }, + new[] { new UnsLineProjection("l1", "a1", "line1") }, + new[] { new EquipmentNode("EQ-1", "station1", "l1") }, + Array.Empty(), + Array.Empty()) + { + RawTags = new[] + { + new RawTagPlan("t1", "Plant/Galaxy/Oven7/TempHi", "Plant/Galaxy/Oven7", "drv-gx", "TempHi", "Boolean", + Writable: false, Alarm: new EquipmentTagAlarmInfo("LimitAlarm", 700), + ReferencingEquipmentPaths: new[] { "filling/line1/station1" }, + NativeAlarmReference: "Oven_007.TempHiAlarm"), + new RawTagPlan("t2", "Plant/Modbus/plc/Hi", "Plant/Modbus/plc", "drv-mb", "Hi", "Boolean", + Writable: false, Alarm: new EquipmentTagAlarmInfo("OffNormalAlarm", 500), + ReferencingEquipmentPaths: Array.Empty()), + }, + }; + + NewApplier(sink).MaterialiseRawSubtree(composition); + + sink.HistoryBindings.ShouldBe(new[] { ("Plant/Galaxy/Oven7/TempHi", AddressSpaceRealm.Raw, "Oven_007.TempHiAlarm") }); + sink.Calls.ShouldBe(new[] { "condition:Plant/Galaxy/Oven7/TempHi", "bind:Plant/Galaxy/Oven7/TempHi", + "wire:Plant/Galaxy/Oven7/TempHi", "condition:Plant/Modbus/plc/Hi" }); + } + /// /// Wave C review M2 — composer↔applier PATH-AGREEMENT (the tripwire for the latent DisplayName==Name /// coupling). The composer builds a native alarm's ReferencingEquipmentPaths from the @@ -489,11 +521,26 @@ public sealed class AddressSpaceApplierRawUnsTests public void EnsureVariable(string variableNodeId, string? parentFolderNodeId, string displayName, string dataType, bool writable, AddressSpaceRealm realm, string? historianTagname = null, bool isArray = false, uint? arrayLength = null, bool refuseArrayOnScalar = false) => Variables.Add((variableNodeId, parentFolderNodeId, writable, historianTagname, realm, isArray, arrayLength)); + public List<(string AlarmNodeId, AddressSpaceRealm AlarmRealm, string Reference)> HistoryBindings { get; } = new(); + public List Calls { get; } = new(); + public void MaterialiseAlarmCondition(string alarmNodeId, string equipmentNodeId, string displayName, string alarmType, int severity, AddressSpaceRealm realm, bool isNative = false) - => Conditions.Add((alarmNodeId, equipmentNodeId, isNative, realm)); + { + Conditions.Add((alarmNodeId, equipmentNodeId, isNative, realm)); + Calls.Add($"condition:{alarmNodeId}"); + } public void WireAlarmNotifiers(string alarmNodeId, AddressSpaceRealm alarmRealm, IReadOnlyList notifierFolderNodeIds, AddressSpaceRealm notifierFolderRealm) - => NotifierWirings.Add((alarmNodeId, alarmRealm, notifierFolderNodeIds, notifierFolderRealm)); + { + NotifierWirings.Add((alarmNodeId, alarmRealm, notifierFolderNodeIds, notifierFolderRealm)); + Calls.Add($"wire:{alarmNodeId}"); + } + + public void BindNativeAlarmHistory(string alarmNodeId, AddressSpaceRealm alarmRealm, string alarmReference) + { + HistoryBindings.Add((alarmNodeId, alarmRealm, alarmReference)); + Calls.Add($"bind:{alarmNodeId}"); + } public void AddReference(string sourceNodeId, AddressSpaceRealm sourceRealm, string targetNodeId, AddressSpaceRealm targetRealm, string referenceType = "Organizes") => References.Add((sourceNodeId, sourceRealm, targetNodeId, targetRealm, referenceType)); diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/AddressSpaceComposerDualNamespaceTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/AddressSpaceComposerDualNamespaceTests.cs index a7baea9a3..0230c7e74 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/AddressSpaceComposerDualNamespaceTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/AddressSpaceComposerDualNamespaceTests.cs @@ -316,6 +316,42 @@ public sealed class AddressSpaceComposerDualNamespaceTests c.UnsReferenceVariables.Single(v => v.UnsTagReferenceId == "r3").BackingScalarRefusesArrayValue.ShouldBeFalse(); } + /// Gitea #886: a Galaxy tag that carries an alarm names its AVEVA alarm reference (its attributeRef), so + /// the folders its condition fires on can read the history AVEVA stores under that object. A Galaxy value tag, and an + /// alarm tag of any other driver (even with an attributeRef-shaped key), name none. + [Fact] + public void A_galaxy_alarm_tag_names_its_aveva_alarm_reference_and_no_other_tag_does() + { + var folder = new RawFolder { RawFolderId = "f", ClusterId = "c1", Name = "Plant" }; + var gx = new DriverInstance { DriverInstanceId = "drv-gx", ClusterId = "c1", RawFolderId = "f", Name = "Galaxy", DriverType = "GalaxyMxGateway", DriverConfig = "{}" }; + var mb = new DriverInstance { DriverInstanceId = "drv-mb", ClusterId = "c1", RawFolderId = "f", Name = "Mb", DriverType = "Modbus", DriverConfig = "{}" }; + var oven = new Device { DeviceId = "d-gx", DriverInstanceId = "drv-gx", Name = "Oven7", DeviceConfig = "{}" }; + var plc = new Device { DeviceId = "d-mb", DriverInstanceId = "drv-mb", Name = "plc", DeviceConfig = "{}" }; + const string alarm = "\"alarm\":{\"alarmType\":\"LimitAlarm\",\"severity\":700}"; + var tags = new[] + { + new Tag { TagId = "t-hi", DeviceId = "d-gx", Name = "TempHi", DataType = "Boolean", AccessLevel = TagAccessLevel.Read, + TagConfig = $"{{\"attributeRef\":\"Oven_007.TempHiAlarm\",{alarm}}}" }, + new Tag { TagId = "t-temp", DeviceId = "d-gx", Name = "Temp", DataType = "Float", AccessLevel = TagAccessLevel.Read, + TagConfig = "{\"attributeRef\":\"Oven_007.Temp\"}" }, + new Tag { TagId = "t-mb", DeviceId = "d-mb", Name = "Hi", DataType = "Boolean", AccessLevel = TagAccessLevel.Read, + TagConfig = $"{{\"region\":\"Coils\",\"address\":0,\"dataType\":\"Bool\",\"attributeRef\":\"Oven_007.X\",{alarm}}}" }, + }; + + var c = AddressSpaceComposer.Compose( + Array.Empty(), Array.Empty(), Array.Empty(), new[] { gx, mb }, Array.Empty(), + tags: tags, rawFolders: new[] { folder }, devices: new[] { oven, plc }); + + c.RawTags.Single(t => t.TagId == "t-hi").NativeAlarmReference.ShouldBe("Oven_007.TempHiAlarm"); + c.RawTags.Single(t => t.TagId == "t-temp").NativeAlarmReference.ShouldBeNull(); + c.RawTags.Single(t => t.TagId == "t-mb").Alarm.ShouldNotBeNull(); + c.RawTags.Single(t => t.TagId == "t-mb").NativeAlarmReference.ShouldBeNull(); + + // The reference is part of the plan's identity: a re-pointed attributeRef is a change, not a no-op redeploy. + var hi = c.RawTags.Single(t => t.TagId == "t-hi"); + hi.Equals(hi with { NativeAlarmReference = "Oven_008.TempHiAlarm" }).ShouldBeFalse(); + } + /// #610 review MINOR-1 — the composer hands the table the owning driver's DriverConfig AND the /// tag's own device's DeviceConfig, so a Modbus write cap configured at either level keeps an array that /// can never fit read-only — on the raw node and its UNS mirror alike. The same tag under an uncapped device is diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/EventSourceFanOutTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/EventSourceFanOutTests.cs new file mode 100644 index 000000000..39c64832e --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/EventSourceFanOutTests.cs @@ -0,0 +1,202 @@ +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; +using HistorianRead = ZB.MOM.WW.OtOpcUa.Core.Abstractions.HistoryReadResult; + +namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests; + +/// +/// Gitea #886 — a folder's several event sources read as one: merged in (EventTimeUtc, EventId) order, the cap +/// applied to the merge, and "more" never claimed for a stretch a truncated source has not read yet. +/// +public sealed class EventSourceFanOutTests +{ + private static readonly DateTime T0 = new(2026, 11, 12, 10, 0, 0, DateTimeKind.Utc); + + private static HistoricalEvent Ev(string id, string source, int second) => + new(id, source, T0.AddSeconds(second), T0.AddSeconds(second), null, 1); + + [Fact] + public async Task One_source_is_read_straight_through() + { + var backend = new StoredEvents(("A", Ev("a1", "A", 1)), ("A", Ev("a2", "A", 2))); + + var r = await EventSourceFanOut.ReadAsync(backend, ["A"], T0, T0.AddMinutes(1), 10, reverse: false, TestContext.Current.CancellationToken); + + r.Events.Select(e => e.EventId).ShouldBe(["a1", "a2"]); + backend.Requests.ShouldBe(["A"]); + } + + [Fact] + public async Task Two_sources_merge_in_time_then_source_then_arrival_order_and_drop_a_duplicate() + { + // Ties keep each source's own order (z-tie before y-tie: A's arrival), and A's before B's (list order) — never + // re-sorted by EventId, which could move an unread tie in front of one already sent. + var backend = new StoredEvents( + ("A", Ev("a1", "A", 1)), ("A", Ev("a3", "A.X", 3)), ("A", Ev("z-tie", "A", 5)), ("A", Ev("y-tie", "A", 5)), + ("B", Ev("b2", "B.Y", 2)), ("B", Ev("a-tie", "B", 5)), ("B", Ev("a1", "A", 1))); + + var r = await EventSourceFanOut.ReadAsync(backend, ["A", "B"], T0, T0.AddMinutes(1), 10, reverse: false, TestContext.Current.CancellationToken); + + r.Events.Select(e => e.EventId).ShouldBe(["a1", "b2", "a3", "z-tie", "y-tie", "a-tie"]); + r.ContinuationPoint.ShouldBeNull(); + backend.Requests.ShouldBe(["A", "B"]); + } + + [Fact] + public async Task A_reverse_read_is_the_forward_merge_backwards() + { + var backend = new StoredEvents( + ("A", Ev("a1", "A", 1)), ("A", Ev("z-tie", "A", 5)), ("A", Ev("y-tie", "A", 5)), + ("B", Ev("b2", "B", 2)), ("B", Ev("a-tie", "B", 5))); + var ct = TestContext.Current.CancellationToken; + + var forward = await EventSourceFanOut.ReadAsync(backend, ["A", "B"], T0, T0.AddMinutes(1), 10, reverse: false, ct); + var reverse = await EventSourceFanOut.ReadAsync(backend, ["A", "B"], T0.AddMinutes(1), T0, 10, reverse: true, ct); + + // What the server's tie-cluster over-fetch relies on: a forward read of a cluster, reversed, is the reverse order. + reverse.Events.Select(e => e.EventId).ShouldBe(forward.Events.Select(e => e.EventId).Reverse()); + reverse.Events.Select(e => e.EventId).ShouldBe(["a-tie", "y-tie", "z-tie", "b2", "a1"]); + } + + [Fact] + public async Task The_cap_applies_to_the_merge_and_flags_more() + { + var backend = new StoredEvents(("A", Ev("a1", "A", 1)), ("A", Ev("a3", "A", 3)), ("B", Ev("b2", "B", 2))); + + var r = await EventSourceFanOut.ReadAsync(backend, ["A", "B"], T0, T0.AddMinutes(1), 2, reverse: false, TestContext.Current.CancellationToken); + + r.Events.Select(e => e.EventId).ShouldBe(["a1", "b2"]); + r.ContinuationPoint.ShouldNotBeNull(); + } + + [Fact] + public async Task A_truncated_source_bounds_the_page_at_the_time_it_stopped() + { + // A default-capped read (maxEvents 0): A stops after two of its three events (at 2); B returns all of its own. + // Unbounded, the page would run on to b4 although A's a3 (at 3) is still unread, so the next page would start + // after it. The page ends at A's stop, with B's tie at 2 left out because it follows A's (unread) ties there. + var backend = new StoredEvents( + ("A", Ev("a1", "A", 1)), ("A", Ev("a2", "A", 2)), ("A", Ev("a3", "A", 3)), + ("B", Ev("b2", "B", 2)), ("B", Ev("b4", "B", 4))) + { DefaultCaps = { ["A"] = 2 } }; + + var r = await EventSourceFanOut.ReadAsync(backend, ["A", "B"], T0, T0.AddMinutes(1), 0, reverse: false, TestContext.Current.CancellationToken); + + r.Events.Select(e => e.EventId).ShouldBe(["a1", "a2"]); + r.ContinuationPoint.ShouldNotBeNull(); + } + + [Fact] + public async Task At_the_bound_a_complete_source_ahead_of_the_truncated_one_keeps_its_ties() + { + var backend = new StoredEvents( + ("A", Ev("a1", "A", 1)), ("A", Ev("a2", "A", 2)), ("A", Ev("a3", "A", 3)), + ("B", Ev("b2", "B", 2)), ("B", Ev("b4", "B", 4))) + { DefaultCaps = { ["A"] = 2 } }; + + // B is listed first, so at 2 its tie precedes A's in merge order and is complete: it stays. + var r = await EventSourceFanOut.ReadAsync(backend, ["B", "A"], T0, T0.AddMinutes(1), 0, reverse: false, TestContext.Current.CancellationToken); + + r.Events.Select(e => e.EventId).ShouldBe(["a1", "b2", "a2"]); + r.ContinuationPoint.ShouldNotBeNull(); + } + + [Fact] + public async Task A_truncated_source_bounds_a_reverse_page_at_the_oldest_time_it_returned() + { + var backend = new StoredEvents( + ("A", Ev("a3", "A", 3)), ("A", Ev("a4", "A", 4)), ("A", Ev("a5", "A", 5)), + ("B", Ev("b4", "B", 4)), ("B", Ev("b1", "B", 1))) + { DefaultCaps = { ["A"] = 2 } }; + + var r = await EventSourceFanOut.ReadAsync(backend, ["A", "B"], T0.AddMinutes(1), T0, 0, reverse: true, TestContext.Current.CancellationToken); + + // A returned 5, 4 and stopped at 4. Reverse merge order at 4 visits B (listed later) first, then A: b4 is + // complete and kept, A's a4 closes the page, and b1 (older than A's unread a3) is left for the next page. + r.Events.Select(e => e.EventId).ShouldBe(["a5", "b4", "a4"]); + r.ContinuationPoint.ShouldNotBeNull(); + } + + [Fact] + public async Task Paging_the_merge_from_each_last_event_walks_every_event_once() + { + // The server's resume rule: read again from the last event's time INCLUSIVE and skip the ties already sent. + var stored = new List<(string, HistoricalEvent)>(); + for (var i = 0; i < 9; i++) stored.Add(("A", Ev($"a{i}", "A", i))); + for (var i = 0; i < 9; i += 2) stored.Add(("B", Ev($"b{i}", "B.X", i))); + stored.Add(("B", Ev("b-tie", "B", 4))); + var backend = new StoredEvents([.. stored]); + var whole = (await EventSourceFanOut.ReadAsync(backend, ["A", "B"], T0, T0.AddMinutes(1), 100, false, TestContext.Current.CancellationToken)) + .Events.Select(e => e.EventId).ToList(); + + foreach (var cap in new[] { 1, 2, 3, 5 }) + { + var got = new List(); + var start = T0; + var skip = 0; + for (var guard = 0; guard < 100; guard++) + { + var page = await EventSourceFanOut.ReadAsync(backend, ["A", "B"], start, T0.AddMinutes(1), cap + skip, false, TestContext.Current.CancellationToken); + var events = page.Events.SkipWhile((e, i) => i < skip && e.EventTimeUtc == start).Take(cap).ToList(); + got.AddRange(events.Select(e => e.EventId)); + var more = page.ContinuationPoint is not null || page.Events.Count - Math.Min(skip, page.Events.Count) > cap; + if (!more) break; + if (events.Count == 0) { skip += cap; continue; } + var boundary = events[^1].EventTimeUtc; + var trailing = events.AsEnumerable().Reverse().TakeWhile(e => e.EventTimeUtc == boundary).Count(); + skip = boundary == start ? skip + trailing : trailing; + start = boundary; + } + + got.ShouldBe(whole, $"cap {cap}"); + } + } + + /// A backend holding events per source; a read returns that source's events in the window, forward or reverse, + /// cut to the cap with a continuation point when more remain. + private sealed class StoredEvents(params (string Source, HistoricalEvent Event)[] stored) : IHistorianDataSource + { + public List Requests { get; } = []; + + /// A source's own cap for a default-capped read (maxEvents <= 0); uncapped when absent. + public Dictionary DefaultCaps { get; } = []; + + public bool SupportsReverseEvents => true; + + public Task ReadEventsAsync( + string? sourceName, DateTime startUtc, DateTime endUtc, int maxEvents, bool reverse, CancellationToken cancellationToken) + { + Requests.Add(sourceName); + var (lo, hi) = reverse ? (endUtc, startUtc) : (startUtc, endUtc); + var rows = stored.Where(s => s.Source == sourceName) + .Select(s => s.Event) + .Where(e => reverse ? e.EventTimeUtc > lo && e.EventTimeUtc <= hi : e.EventTimeUtc >= lo && e.EventTimeUtc < hi) + .OrderBy(e => e.EventTimeUtc) // stable: ties keep storage order, as the gateway's do + .ToList(); + if (reverse) rows.Reverse(); + var cap = maxEvents > 0 ? maxEvents : DefaultCaps.GetValueOrDefault(sourceName ?? string.Empty, 0); + var more = cap > 0 && rows.Count > cap; + return Task.FromResult(new HistoricalEventsResult(more ? rows.Take(cap).ToList() : rows, more ? [0x01] : null)); + } + + public Task ReadEventsAsync( + string? sourceName, DateTime startUtc, DateTime endUtc, int maxEvents, CancellationToken cancellationToken) => + ReadEventsAsync(sourceName, startUtc, endUtc, maxEvents, reverse: false, cancellationToken); + + public Task ReadRawAsync(string fullReference, DateTime startUtc, DateTime endUtc, uint maxValuesPerNode, + CancellationToken cancellationToken) => throw new NotSupportedException(); + + public Task ReadProcessedAsync(string fullReference, DateTime startUtc, DateTime endUtc, TimeSpan interval, + HistoryAggregateType aggregate, CancellationToken cancellationToken) => throw new NotSupportedException(); + + public Task ReadAtTimeAsync(string fullReference, IReadOnlyList timestampsUtc, + CancellationToken cancellationToken) => throw new NotSupportedException(); + + public HistorianHealthSnapshot GetHealthSnapshot() => NullHistorianDataSource.Instance.GetHealthSnapshot(); + + public void Dispose() + { + } + } +}