Merge b640-619alm: OpcUaClient native alarms route to RawPath conditions + ConditionRefresh; docker-dev opc-plc seed (#619)

Claude-Session: https://claude.ai/code/session_01Ef2ozxS7QPanGT8uoATbTe
This commit is contained in:
Joseph Doherty
2026-10-08 17:26:47 -04:00
5 changed files with 870 additions and 19 deletions
+30
View File
@@ -201,3 +201,33 @@ IF NOT EXISTS (SELECT 1 FROM dbo.Tag WHERE TagId = 'MAIN-modbus-hr200')
VALUES ('MAIN-modbus-hr200', 'MAIN-modbus-dev', NULL, 'HR200', 'UInt16', 'ReadWrite', 1, VALUES ('MAIN-modbus-hr200', 'MAIN-modbus-dev', NULL, 'HR200', 'UInt16', 'ReadWrite', 1,
N'{ "region": "HoldingRegisters", "address": 200, "dataType": "UInt16", "writable": true }'); N'{ "region": "HoldingRegisters", "address": 200, "dataType": "UInt16", "writable": true }');
SELECT DriverInstanceId, ClusterId, DriverType, Enabled FROM dbo.DriverInstance WHERE ClusterId='MAIN'; SELECT DriverInstanceId, ClusterId, DriverType, Enabled FROM dbo.DriverInstance WHERE ClusterId='MAIN';
------------------------------------------------------------------------------
-- #619: OpcUaClient driver against the opc-plc fixture (10.100.0.35:50000, started by
-- tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.IntegrationTests/Docker with --alm).
-- Tag identity under v3 is the RawPath; the upstream node is authored in TagConfig.nodeId.
-- * FastUInt1 (ns=3;s=FastUInt1) is the always-moving VALUE control: proves the driver is connected and
-- the RawPath -> nodeId table resolves before any alarm assertion is trusted.
-- * The ALARM tag's nodeId must be the SourceNode opc-plc stamps on its condition events - the driver
-- matches events on it. It is NOT guessable (and opc-plc was not running when this was authored), so
-- it is supplied at seed time: browse opc-plc (Client.CLI `browse -r -d 4`, or `alarms --refresh`
-- which prints each event's source) and set @OpcPlcAlarmSourceNodeId below. Empty = no alarm tag.
-- Idempotent.
------------------------------------------------------------------------------
DECLARE @OpcPlcAlarmSourceNodeId nvarchar(400) = N''; -- e.g. N'ns=3;s=<source from opc-plc>'
IF NOT EXISTS (SELECT 1 FROM dbo.DriverInstance WHERE DriverInstanceId = 'MAIN-opcua')
INSERT dbo.DriverInstance(DriverInstanceId, ClusterId, RawFolderId, Name, DriverType, DriverConfig, Enabled)
VALUES ('MAIN-opcua', 'MAIN', NULL, 'opcplc', 'OpcUaClient',
N'{ "SecurityPolicy": "None", "SecurityMode": "None", "AuthType": "Anonymous", "AutoAcceptCertificates": true }', 1);
IF NOT EXISTS (SELECT 1 FROM dbo.Device WHERE DeviceId = 'MAIN-opcua-dev')
INSERT dbo.Device(DeviceId, DriverInstanceId, Name, Enabled, DeviceConfig)
VALUES ('MAIN-opcua-dev', 'MAIN-opcua', 'plc', 1, N'{ "EndpointUrl": "opc.tcp://10.100.0.35:50000" }');
IF NOT EXISTS (SELECT 1 FROM dbo.Tag WHERE TagId = 'MAIN-opcua-fastuint1')
INSERT dbo.Tag(TagId, DeviceId, TagGroupId, Name, DataType, AccessLevel, WriteIdempotent, TagConfig)
VALUES ('MAIN-opcua-fastuint1', 'MAIN-opcua-dev', NULL, 'FastUInt1', 'UInt32', 'Read', 0,
N'{ "nodeId": "ns=3;s=FastUInt1" }');
IF LEN(@OpcPlcAlarmSourceNodeId) > 0 AND NOT EXISTS (SELECT 1 FROM dbo.Tag WHERE TagId = 'MAIN-opcua-alarm1')
INSERT dbo.Tag(TagId, DeviceId, TagGroupId, Name, DataType, AccessLevel, WriteIdempotent, TagConfig)
VALUES ('MAIN-opcua-alarm1', 'MAIN-opcua-dev', NULL, 'Alarm1', 'Boolean', 'Read', 0,
N'{ "nodeId": "' + @OpcPlcAlarmSourceNodeId + N'", "alarm": { "alarmType": "AlarmCondition", "severity": 500 } }');
SELECT DriverInstanceId, ClusterId, DriverType, Enabled FROM dbo.DriverInstance WHERE DriverInstanceId = 'MAIN-opcua';
+18
View File
@@ -91,6 +91,24 @@ rather than surfacing as a runtime surprise.
| `IAlarmSource` | `SubscribeAlarmsAsync` (EventFilter) + `AcknowledgeAsync` | Subscribes to upstream alarm/condition events and forwards acks | | `IAlarmSource` | `SubscribeAlarmsAsync` (EventFilter) + `AcknowledgeAsync` | Subscribes to upstream alarm/condition events and forwards acks |
| `IHistoryProvider` | `ReadRawAsync` / `ReadProcessedAsync` / `ReadAtTimeAsync` → `Session.HistoryReadAsync` | **Unique to this driver** — passthrough history read against the upstream server | | `IHistoryProvider` | `ReadRawAsync` / `ReadProcessedAsync` / `ReadAtTimeAsync` → `Session.HistoryReadAsync` | **Unique to this driver** — passthrough history read against the upstream server |
> **Native alarms (#619).** An alarm raw tag is `TagConfig` `{ "nodeId": "<upstream SourceNode>", "alarm": {...} }`.
> The host subscribes with RawPaths and routes on (driver, RawPath), so the driver matches each
> event's `ConditionId` (then its `SourceNode`) against the tag's authored `nodeId` and reports the **RawPath** as
> `ConditionId`. Author `nodeId` as the upstream **condition** node for a 1:1 mapping (required when one source carries
> several conditions, e.g. Hi + HiHi); authoring the **source** node folds every condition of that source into one
> tag. `Acknowledge` is called on the condition node of the last event seen for the RawPath **with that event's
> EventId** (Part 9 §5.7.3). The event's `ActiveState/Id` + `AckedState/Id` are selected and turned into the
> `Raise`/`Acknowledge`/`Clear`/`Retrigger` kinds the host projector consumes (without a kind no upstream alarm could ever
> show active); the kinds are emitted from the state **last told to the host** per RawPath, so an identical re-delivery
> emits nothing (no re-raise, which would un-ack). Condition **branch** events (non-null `BranchId`) are ignored.
> **`ConditionRefresh`** is called after the event item is created, after every in-place reconnect, and on a
> `RefreshRequired` event, so a condition already active upstream is reported at subscribe; `RefreshStart`/`RefreshEnd`
> markers are dropped. A server without it (`BadNotSupported`/`BadMethodInvalid`/…) is logged once and tolerated.
> Last-told state survives a re-initialise (the host re-subscribes the same set over nodes it did not rebuild) and is
> forgotten when the subscribed set changes (the host rebuilt, recreating the nodes inactive). Index keys are server-stable (`nsu=`), so a namespace reorder across a reconnect keeps
> matching. docker-dev seeds an opc-plc driver (`MAIN-opcua`); the alarm tag is
> created only when `@OpcPlcAlarmSourceNodeId` is set in `docker-dev/seed/seed-clusters.sql`.
> This driver does **not** implement `IRediscoverable` — there is no > This driver does **not** implement `IRediscoverable` — there is no
> push-driven rediscovery signal from a remote OPC UA server in this driver. > push-driven rediscovery signal from a remote OPC UA server in this driver.
> `IHistoryProvider` is implemented by no other driver; history reads for every > `IHistoryProvider` is implemented by no other driver; history reads for every
@@ -92,6 +92,29 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
// ---- IAlarmSource state ---- // ---- IAlarmSource state ----
private readonly System.Collections.Concurrent.ConcurrentDictionary<long, RemoteAlarmSubscription> _alarmSubscriptions = new(); private readonly System.Collections.Concurrent.ConcurrentDictionary<long, RemoteAlarmSubscription> _alarmSubscriptions = new();
/// <summary>#619: RawPath → the upstream condition NodeId last seen for it. The host acknowledges by
/// RawPath (its only identity for the condition) but Part 9 Acknowledge must be called on the CONDITION
/// node, which differs from the authored source node.</summary>
private readonly System.Collections.Concurrent.ConcurrentDictionary<string, AlarmAckTarget> _alarmConditionByRawPath =
new(StringComparer.Ordinal);
/// <summary>#619 review: the (Active, Acked) state last told to the host per reported condition id, from
/// which <see cref="AlarmTransitions"/> picks the Kind sequence that moves the host's projector to the
/// upstream event's state.</summary>
private readonly Dictionary<string, AlarmToldState> _alarmStateByRawPath = new(StringComparer.Ordinal);
/// <summary>#619 re-review: serialises the read-compute-emit over <see cref="_alarmStateByRawPath"/>. Event
/// notifications arrive on SDK publish threads (and, across a resubscribe, from two subscriptions at once), so
/// a lock-free prior/target exchange could emit two kind sequences computed from the same prior.</summary>
private readonly object _alarmStateLock = new();
/// <summary>#619 re-review: the ref set the last <see cref="SubscribeAlarmsAsync"/> carried. Survives
/// <see cref="ShutdownAsync"/> deliberately — see <see cref="BeginAlarmSubscription"/>.</summary>
private HashSet<string>? _lastAlarmRefSet;
/// <summary>Latched once the upstream server has refused ConditionRefresh as unsupported (logged once).</summary>
private int _conditionRefreshUnsupportedLogged;
private long _nextAlarmSubscriptionId; private long _nextAlarmSubscriptionId;
/// <summary>Occurs when an alarm event is received from the remote OPC UA server.</summary> /// <summary>Occurs when an alarm event is received from the remote OPC UA server.</summary>
@@ -879,6 +902,11 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
catch { /* best-effort */ } catch { /* best-effort */ }
} }
_alarmSubscriptions.Clear(); _alarmSubscriptions.Clear();
// #619 re-review: the last-told alarm states (and ack targets) deliberately SURVIVE a shutdown. A
// ReinitializeAsync (comms-loss watchdog / Force Reconnect) is followed by the host re-subscribing the
// same set over condition nodes it did NOT rebuild; forgetting here would make the post-subscribe
// ConditionRefresh re-raise every active+acked alarm (an un-ack flicker). BeginAlarmSubscription forgets
// them when the set changes, which is when the host did rebuild.
await ReleaseSessionAsync(cancellationToken).ConfigureAwait(false); await ReleaseSessionAsync(cancellationToken).ConfigureAwait(false);
_namespaceMap = null; _namespaceMap = null;
@@ -1621,6 +1649,9 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
private const int AlarmFieldSeverity = 4; private const int AlarmFieldSeverity = 4;
private const int AlarmFieldTime = 5; private const int AlarmFieldTime = 5;
private const int AlarmFieldConditionId = 6; private const int AlarmFieldConditionId = 6;
private const int AlarmFieldActiveState = 7;
private const int AlarmFieldAckedState = 8;
private const int AlarmFieldBranchId = 9;
/// <inheritdoc /> /// <inheritdoc />
public async Task<IAlarmSubscriptionHandle> SubscribeAlarmsAsync( public async Task<IAlarmSubscriptionHandle> SubscribeAlarmsAsync(
@@ -1635,11 +1666,24 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
// Pre-resolve the source-node filter set so the per-event notification handler can // Pre-resolve the source-node filter set so the per-event notification handler can
// match in O(1) without re-parsing on every event. // match in O(1) without re-parsing on every event.
var sourceFilter = new HashSet<string>(sourceNodeIds, StringComparer.Ordinal); // #619: under v3 the host passes RawPaths, never upstream node ids. Translate each to the upstream
// source id its tag authored (TagConfig.nodeId) so the per-event match compares like with like.
// Keys are server-stable (AlarmIndexKey) so a namespace reorder across an in-place SDK reconnect —
// which transfers this subscription without re-running SubscribeAlarmsAsync — cannot orphan them.
var sessionForResolve = Session;
var mapForResolve = _namespaceMap;
var sourceFilter = BuildAlarmSourceIndex(sourceNodeIds, raw =>
{
if (sessionForResolve is null) return null;
if (!TryResolveTagNode(sessionForResolve, raw, out var live)
&& !NamespaceMap.TryResolve(sessionForResolve, raw, out live)) return null;
return AlarmIndexKey(mapForResolve, live);
});
BeginAlarmSubscription(sourceNodeIds);
var subscription = new Subscription(telemetry: null!, new SubscriptionOptions var subscription = new Subscription(telemetry: null!, new SubscriptionOptions
{ {
DisplayName = $"opcua-alarm-sub-{id}", DisplayName = $"{AlarmSubscriptionNamePrefix}{id}",
PublishingInterval = 500, // 500ms — alarms don't need fast polling; the server pushes PublishingInterval = 500, // 500ms — alarms don't need fast polling; the server pushes
KeepAliveCount = _options.SubscriptionKeepAliveCount, KeepAliveCount = _options.SubscriptionKeepAliveCount,
LifetimeCount = _options.SubscriptionLifetimeCount, LifetimeCount = _options.SubscriptionLifetimeCount,
@@ -1674,6 +1718,28 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
BrowsePath = [], // empty path = the condition node itself BrowsePath = [], // empty path = the condition node itself
AttributeId = Attributes.NodeId, AttributeId = Attributes.NodeId,
}); });
// #619 review: the host derives Active/Acked solely from AlarmEventArgs.Kind, so the condition state
// must be selected. TWO-element browse paths — a single "ActiveState/Id" QualifiedName names nothing.
filter.SelectClauses.Add(new SimpleAttributeOperand
{
TypeDefinitionId = ObjectTypeIds.AlarmConditionType,
BrowsePath = [new QualifiedName(BrowseNames.ActiveState), new QualifiedName(BrowseNames.Id)],
AttributeId = Attributes.Value,
});
filter.SelectClauses.Add(new SimpleAttributeOperand
{
TypeDefinitionId = ObjectTypeIds.AcknowledgeableConditionType,
BrowsePath = [new QualifiedName(BrowseNames.AckedState), new QualifiedName(BrowseNames.Id)],
AttributeId = Attributes.Value,
});
// #619 re-review: BranchId — a non-null one marks a condition BRANCH event (a previous state, with its own
// EventId). Only the trunk (null BranchId) is the condition's current state.
filter.SelectClauses.Add(new SimpleAttributeOperand
{
TypeDefinitionId = ObjectTypeIds.ConditionType,
BrowsePath = [new QualifiedName(BrowseNames.BranchId)],
AttributeId = Attributes.Value,
});
await _gate.WaitAsync(cancellationToken).ConfigureAwait(false); await _gate.WaitAsync(cancellationToken).ConfigureAwait(false);
try try
@@ -1710,6 +1776,11 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
await subscription.CreateItemsAsync(cancellationToken).ConfigureAwait(false); await subscription.CreateItemsAsync(cancellationToken).ConfigureAwait(false);
_alarmSubscriptions[id] = new RemoteAlarmSubscription(subscription, handle, eventItem, notifHandler); _alarmSubscriptions[id] = new RemoteAlarmSubscription(subscription, handle, eventItem, notifHandler);
// #619: a new event item only reports FUTURE transitions — a condition already active upstream would
// stay invisible until it next changed. ConditionRefresh re-delivers every retained condition's
// current state through the normal path (deduplicated against the last-told state).
await TryConditionRefreshAsync(subscription, cancellationToken).ConfigureAwait(false);
} }
finally { _gate.Release(); } finally { _gate.Release(); }
@@ -1765,15 +1836,30 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
var callRequests = new CallMethodRequestCollection(); var callRequests = new CallMethodRequestCollection();
foreach (var ack in acknowledgements) foreach (var ack in acknowledgements)
{ {
// ConditionId is an upstream condition-node id captured from the alarm event (not a // #619: the host acks by RawPath; Acknowledge must hit the upstream CONDITION node.
// tag RawPath), so it re-binds directly against the session via the static resolve. // The remembered target pairs the condition node with the EventId of the event that named it,
if (!NamespaceMap.TryResolve(session, ack.ConditionId, out var conditionId)) continue; // so the ack can never mix one condition's node with another's EventId. Nothing seen yet ⇒ the
// tag's authored node (correct when the tag was authored on the condition node), else the id
// verbatim (pre-v3 callers that pass an upstream condition id directly).
NodeId conditionId;
byte[] eventId;
if (_alarmConditionByRawPath.TryGetValue(ack.ConditionId, out var seen))
{
if (!NamespaceMap.TryResolve(session, seen.ConditionRef, out conditionId)) continue;
eventId = seen.EventId;
}
else
{
if (!TryResolveTagNode(session, ack.ConditionId, out conditionId)
&& !NamespaceMap.TryResolve(session, ack.ConditionId, out conditionId)) continue;
eventId = Array.Empty<byte>();
}
callRequests.Add(new CallMethodRequest callRequests.Add(new CallMethodRequest
{ {
ObjectId = conditionId, ObjectId = conditionId,
MethodId = MethodIds.AcknowledgeableConditionType_Acknowledge, MethodId = MethodIds.AcknowledgeableConditionType_Acknowledge,
InputArguments = [ InputArguments = [
new Variant(Array.Empty<byte>()), // EventId — server-side best-effort; empty resolves to 'most recent' new Variant(eventId), // Part 9 §5.7.3: the EventId being acknowledged (empty only when none seen)
new Variant(new LocalizedText(ack.Comment ?? string.Empty)), new Variant(new LocalizedText(ack.Comment ?? string.Empty)),
], ],
}); });
@@ -1831,32 +1917,339 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
private void OnEventNotification( private void OnEventNotification(
OpcUaAlarmSubscriptionHandle handle, OpcUaAlarmSubscriptionHandle handle,
HashSet<string> sourceFilter, IReadOnlyDictionary<string, string> sourceFilter,
MonitoredItem item, MonitoredItem item,
MonitoredItemNotificationEventArgs args) MonitoredItemNotificationEventArgs args)
{ {
if (args.NotificationValue is not EventFieldList efl) return; if (args.NotificationValue is not EventFieldList efl) return;
if (efl.EventFields.Count <= AlarmFieldConditionId) return; RouteAlarmEvent(efl.EventFields, sourceFilter, handle);
}
var sourceNode = efl.EventFields[AlarmFieldSourceNode].Value?.ToString() ?? string.Empty; /// <summary>
if (sourceFilter.Count > 0 && !sourceFilter.Contains(sourceNode)) return; /// #619 re-review: open (or re-open) the host's alarm subscription over <paramref name="sourceNodeIds"/>.
/// A CHANGED set means the host rebuilt its address space (an alarm-bearing delta always does), recreating
/// every condition node inactive+acked — so every last-told state is forgotten and the post-subscribe
/// ConditionRefresh re-reports from that default. The SAME set (a reinit / reconnect re-subscribe) keeps
/// them, so the refresh copy of an already-reported state emits nothing. Ack targets of refs no longer
/// subscribed are pruned either way.
/// </summary>
/// <param name="sourceNodeIds">The refs the host is subscribing.</param>
internal void BeginAlarmSubscription(IReadOnlyList<string> sourceNodeIds)
{
var set = new HashSet<string>(sourceNodeIds, StringComparer.Ordinal);
lock (_alarmStateLock)
{
if (_lastAlarmRefSet is null || !_lastAlarmRefSet.SetEquals(set)) _alarmStateByRawPath.Clear();
_lastAlarmRefSet = set;
foreach (var known in _alarmConditionByRawPath.Keys)
if (!set.Contains(known)) _alarmConditionByRawPath.TryRemove(known, out _);
}
}
var eventType = efl.EventFields[AlarmFieldEventType].Value?.ToString() ?? "BaseEventType"; /// <summary>
var message = (efl.EventFields[AlarmFieldMessage].Value as LocalizedText)?.Text ?? string.Empty; /// #619 — route one upstream event: drop refresh markers (a RefreshRequired asks for a new refresh), project
var severity = efl.EventFields[AlarmFieldSeverity].Value is ushort sev ? sev : (ushort)0; /// it, and emit the kinds that move the host from the state last told to the event's. An event that changes
var time = efl.EventFields[AlarmFieldTime].Value is DateTime t ? t : DateTime.UtcNow; /// nothing the host shows (a ConditionRefresh copy) emits nothing.
var conditionId = efl.EventFields[AlarmFieldConditionId].Value?.ToString() ?? string.Empty; /// </summary>
/// <param name="fields">The event's fields, in select-clause order.</param>
/// <param name="sourceFilter">The subscription's upstream-key → RawPath index.</param>
/// <param name="handle">The subscription handle to stamp.</param>
internal void RouteAlarmEvent(
IReadOnlyList<Variant> fields, IReadOnlyDictionary<string, string> sourceFilter, IAlarmSubscriptionHandle handle)
{
if (IsRefreshRequired(fields))
{
// Part 9 §5.11.7: the server lost events (queue overflow / restart) and asks for a refresh.
_ = Task.Run(() => RefreshAllAlarmSubscriptionsAsync(CancellationToken.None));
return;
}
OnAlarmEvent?.Invoke(this, new AlarmEventArgs( // Projection, the per-RawPath bookkeeping AND the emission run under one lock, so two notifications for
// one condition (two publish threads; two subscriptions across a resubscribe) can never compute their
// kinds from the same prior, nor reach the host out of order. The host's handler only marshals.
lock (_alarmStateLock)
{
var e = ProjectEvent(fields, sourceFilter, handle,
n => AlarmIndexKey(_namespaceMap, n), (raw, target) => _alarmConditionByRawPath[raw] = target,
out var state);
if (e is null) return;
if (state is not { } target)
{
// Not a condition (no ActiveState/AckedState) — a message/severity-only event, as before.
OnAlarmEvent?.Invoke(this, e);
return;
}
// Fresh prior = the host projector's own default (inactive, acked).
var told = new AlarmToldState(target.Active, target.Acked, e.Message, e.OpcUaSeverity);
var prior = _alarmStateByRawPath.TryGetValue(e.ConditionId, out var p)
? p
: new AlarmToldState(false, true, null, null);
_alarmStateByRawPath[e.ConditionId] = told;
var kinds = AlarmTransitions((prior.Active, prior.Acked), target);
if (kinds.Count == 0)
{
// Same state: a message/severity change still reaches the host; an identical copy does not.
if (prior == told) return;
kinds = [AlarmTransitionKind.Unspecified];
}
foreach (var kind in kinds)
OnAlarmEvent?.Invoke(this, e with { Kind = kind });
}
}
/// <summary>#619 re-review — what the host was last told about one condition: its state plus the message and
/// exact severity, so a refresh copy is recognised as a duplicate.</summary>
/// <param name="Active">ActiveState/Id last told.</param>
/// <param name="Acked">AckedState/Id last told.</param>
/// <param name="Message">Message last told (null before any event).</param>
/// <param name="Severity">Exact severity last told (null before any event).</param>
private readonly record struct AlarmToldState(bool Active, bool Acked, string? Message, int? Severity);
/// <summary>True for a <c>RefreshRequiredEventType</c> event (the server asks the client to ConditionRefresh).</summary>
/// <param name="fields">The event's fields.</param>
/// <returns>Whether the event is a RefreshRequired marker.</returns>
internal static bool IsRefreshRequired(IReadOnlyList<Variant> fields) =>
fields.Count > AlarmFieldEventType && fields[AlarmFieldEventType].Value is NodeId t
&& t == ObjectTypeIds.RefreshRequiredEventType;
/// <summary>True for the three ConditionRefresh marker event types — never alarms.</summary>
/// <param name="eventType">The event's EventType.</param>
/// <returns>Whether it is a refresh marker.</returns>
private static bool IsRefreshMarker(object? eventType) =>
eventType is NodeId t
&& (t == ObjectTypeIds.RefreshStartEventType || t == ObjectTypeIds.RefreshEndEventType
|| t == ObjectTypeIds.RefreshRequiredEventType);
/// <summary>A ConditionRefresh refusal meaning "this server does not do refresh" (tolerated, logged once).</summary>
/// <param name="code">The refusal status.</param>
/// <returns>Whether the server simply lacks ConditionRefresh.</returns>
internal static bool IsConditionRefreshUnsupported(StatusCode code) =>
code == StatusCodes.BadNotSupported || code == StatusCodes.BadMethodInvalid
|| code == StatusCodes.BadNotImplemented || code == StatusCodes.BadServiceUnsupported
|| code == StatusCodes.BadNodeIdUnknown;
/// <summary>
/// #619 — call ConditionRefresh for <paramref name="subscription"/> so conditions already active upstream
/// are reported. Never throws: an unsupported server is logged once, any other failure warns.
/// </summary>
/// <param name="subscription">The (created) alarm subscription.</param>
/// <param name="cancellationToken">Cancellation.</param>
/// <returns>A task completing when the refresh call returns.</returns>
private async Task TryConditionRefreshAsync(Subscription subscription, CancellationToken cancellationToken)
{
try
{
await subscription.ConditionRefreshAsync(cancellationToken).ConfigureAwait(false);
}
catch (ServiceResultException ex) when (IsConditionRefreshUnsupported(ex.StatusCode))
{
if (Interlocked.Exchange(ref _conditionRefreshUnsupportedLogged, 1) == 0)
_logger.LogInformation(
"OpcUaClientDriver '{DriverInstanceId}': upstream server does not support ConditionRefresh " +
"({StatusCode}); conditions already active are reported only on their next transition.",
_driverInstanceId, ex.StatusCode);
}
catch (Exception ex) when (ex is not OperationCanceledException || !cancellationToken.IsCancellationRequested)
{
_logger.LogWarning(ex,
"OpcUaClientDriver '{DriverInstanceId}': ConditionRefresh on alarm subscription {SubscriptionId} failed; " +
"conditions already active upstream may not be reported until their next transition.",
_driverInstanceId, subscription.Id);
}
catch (OperationCanceledException)
{
// The caller cancelled; the subscription itself is in place.
}
}
/// <summary>
/// #619 — ConditionRefresh every alarm subscription on the current session: after an in-place reconnect
/// (transferred or SDK-recreated subscriptions — matched by display name, since a recreate clones the
/// <see cref="Subscription"/> objects) and on a RefreshRequired event. Events missed during the outage
/// reach the host through the normal deduplicated path.
/// </summary>
/// <param name="cancellationToken">Cancellation.</param>
/// <returns>A task completing when every refresh call has returned.</returns>
private async Task RefreshAllAlarmSubscriptionsAsync(CancellationToken cancellationToken)
{
if (_alarmSubscriptions.IsEmpty) return;
try
{
await _gate.WaitAsync(cancellationToken).ConfigureAwait(false);
}
catch (ObjectDisposedException) { return; }
try
{
var session = Session;
if (session is null) return;
foreach (var sub in session.Subscriptions.ToList())
{
if (sub.DisplayName?.StartsWith(AlarmSubscriptionNamePrefix, StringComparison.Ordinal) == true && sub.Created)
await TryConditionRefreshAsync(sub, cancellationToken).ConfigureAwait(false);
}
}
catch (Exception ex)
{
_logger.LogWarning(ex, "OpcUaClientDriver '{DriverInstanceId}': alarm ConditionRefresh sweep failed.",
_driverInstanceId);
}
finally
{
try { _gate.Release(); } catch (ObjectDisposedException) { }
}
}
/// <summary>Display-name prefix of every alarm subscription (how a refresh sweep finds them on a session).</summary>
private const string AlarmSubscriptionNamePrefix = "opcua-alarm-sub-";
/// <summary>
/// #619 review — the <see cref="AlarmTransitionKind"/> sequence that moves the host's
/// <c>NativeAlarmProjector</c> (which knows only kinds: Raise ⇒ active+unacked, Acknowledge ⇒ acked,
/// Clear ⇒ inactive keeping acked) from <paramref name="prior"/> to <paramref name="target"/>. A single
/// upstream event can need two (an alarm that went active and was acknowledged between two notifications).
/// No state change ⇒ empty (the caller decides whether a message/severity change warrants an Unspecified).
/// </summary>
/// <param name="prior">The state last told to the host.</param>
/// <param name="target">The upstream event's ActiveState/Id + AckedState/Id.</param>
/// <returns>The kinds to emit, in order.</returns>
internal static IReadOnlyList<AlarmTransitionKind> AlarmTransitions(
(bool Active, bool Acked) prior, (bool Active, bool Acked) target)
{
if (prior == target) return [];
return target switch
{
// Active+unacked: a fresh raise, or a re-arm of an active-acked condition.
(true, false) => [prior.Active ? AlarmTransitionKind.Retrigger : AlarmTransitionKind.Raise],
// Active+acked: raise first when it was inactive (Raise always lands unacked).
(true, true) => prior.Active
? [AlarmTransitionKind.Acknowledge]
: [AlarmTransitionKind.Raise, AlarmTransitionKind.Acknowledge],
// Inactive+acked: Clear keeps the prior ack, so ack after it when that was unacked.
(false, true) => prior.Active
? prior.Acked ? [AlarmTransitionKind.Clear] : [AlarmTransitionKind.Clear, AlarmTransitionKind.Acknowledge]
: [AlarmTransitionKind.Acknowledge],
// Inactive+UNacked (went active and cleared unacknowledged — e.g. first seen by a ConditionRefresh):
// the only kinds that un-ack are Raise/Retrigger, so pass through active, then Clear.
(false, false) => prior.Active
? prior.Acked ? [AlarmTransitionKind.Retrigger, AlarmTransitionKind.Clear] : [AlarmTransitionKind.Clear]
: [AlarmTransitionKind.Raise, AlarmTransitionKind.Clear],
};
}
/// <summary>
/// #619 — index an alarm subscription's source refs by UPSTREAM source id. The host (v3) subscribes with
/// the alarm tags' <b>RawPaths</b>; an upstream event carries its own source NodeId, so the match key is
/// the upstream id the tag authored and the value is the RawPath to report back. A ref
/// <paramref name="resolveUpstream"/> cannot resolve is kept verbatim (direct upstream-id use).
/// </summary>
/// <param name="refs">The refs handed to <c>SubscribeAlarmsAsync</c>.</param>
/// <param name="resolveUpstream">RawPath → serialized live upstream NodeId, or null when unknown.</param>
/// <returns>Upstream source id → ref to report.</returns>
internal static Dictionary<string, string> BuildAlarmSourceIndex(
IReadOnlyList<string> refs, Func<string, string?> resolveUpstream)
{
var index = new Dictionary<string, string>(StringComparer.Ordinal);
foreach (var r in refs)
index[resolveUpstream(r) ?? r] = r;
return index;
}
/// <summary>
/// #619 — project one upstream event into an <see cref="AlarmEventArgs"/>. <c>ConditionId</c> and
/// <c>SourceNodeId</c> are the authored RawPath, because <c>DriverHostActor</c> routes native alarms on
/// (driver, RawPath); the upstream condition node (needed to acknowledge) is handed to
/// <paramref name="rememberCondition"/>. Null when the event's source is not an authored one.
/// </summary>
internal static AlarmEventArgs? ProjectEvent(
IReadOnlyList<Variant> fields,
IReadOnlyDictionary<string, string> sourceIndex,
IAlarmSubscriptionHandle handle,
Func<NodeId, string> keyOf,
Action<string, AlarmAckTarget> rememberCondition) =>
ProjectEvent(fields, sourceIndex, handle, keyOf, rememberCondition, out _);
/// <summary>As the five-argument overload, also returning the event's (ActiveState/Id, AckedState/Id) through
/// <c>state</c> when it is an alarm condition (null otherwise).</summary>
internal static AlarmEventArgs? ProjectEvent(
IReadOnlyList<Variant> fields,
IReadOnlyDictionary<string, string> sourceIndex,
IAlarmSubscriptionHandle handle,
Func<NodeId, string> keyOf,
Action<string, AlarmAckTarget> rememberCondition,
out (bool Active, bool Acked)? state)
{
state = null;
if (fields.Count <= AlarmFieldConditionId) return null;
// ConditionRefresh brackets its re-delivery with RefreshStart/RefreshEnd — markers, never alarms.
if (IsRefreshMarker(fields[AlarmFieldEventType].Value)) return null;
// A condition BRANCH event reports a previous state with its own EventId: projecting it would flip the
// trunk's state, remembering it would make Acknowledge ack the branch.
if (fields.Count > AlarmFieldBranchId && fields[AlarmFieldBranchId].Value is NodeId branch && !NodeId.IsNull(branch))
return null;
if (fields.Count > AlarmFieldAckedState
&& fields[AlarmFieldActiveState].Value is bool active
&& fields[AlarmFieldAckedState].Value is bool acked)
state = (active, acked);
// Pass-through (no authored refs) keeps the raw serialized ids; an authored index compares
// server-stable keys (see AlarmIndexKey), matching the CONDITION node first — the only way to tell
// several conditions on one source apart — and then the source node.
var sourceField = fields[AlarmFieldSourceNode].Value;
var conditionField = fields[AlarmFieldConditionId].Value;
string reported, upstreamCondition;
if (sourceIndex.Count == 0)
{
reported = sourceField?.ToString() ?? string.Empty;
upstreamCondition = conditionField?.ToString() ?? string.Empty;
}
else
{
upstreamCondition = conditionField is NodeId cn && !NodeId.IsNull(cn) ? keyOf(cn) : string.Empty;
var sourceKey = sourceField is NodeId sn && !NodeId.IsNull(sn) ? keyOf(sn) : string.Empty;
if (!(upstreamCondition.Length > 0 && sourceIndex.TryGetValue(upstreamCondition, out reported!))
&& !(sourceKey.Length > 0 && sourceIndex.TryGetValue(sourceKey, out reported!)))
return null;
}
var eventType = fields[AlarmFieldEventType].Value?.ToString() ?? "BaseEventType";
var message = (fields[AlarmFieldMessage].Value as LocalizedText)?.Text ?? string.Empty;
var severity = fields[AlarmFieldSeverity].Value is ushort sev ? sev : (ushort)0;
var time = fields[AlarmFieldTime].Value is DateTime t ? t : DateTime.UtcNow;
if (sourceIndex.Count != 0 && upstreamCondition.Length > 0)
rememberCondition(reported, new AlarmAckTarget(
upstreamCondition,
// A copy: the decoded notification's array is not ours to alias for the ack's lifetime.
(fields[AlarmFieldEventId].Value as byte[])?.ToArray() ?? Array.Empty<byte>()));
return new AlarmEventArgs(
SubscriptionHandle: handle, SubscriptionHandle: handle,
SourceNodeId: sourceNode, SourceNodeId: reported,
ConditionId: conditionId, ConditionId: sourceIndex.Count == 0 ? upstreamCondition : reported,
AlarmType: eventType, AlarmType: eventType,
Message: message, Message: message,
Severity: MapSeverity(severity), Severity: MapSeverity(severity),
SourceTimestampUtc: time)); SourceTimestampUtc: time,
// #626: the upstream severity is already the exact Part 9 value — carry it, don't re-bucket.
OpcUaSeverity: severity == 0 ? null : severity);
} }
/// <summary>#619 — what Acknowledge needs for one RawPath: the upstream condition node (server-stable
/// reference) and the EventId of the event that named it.</summary>
/// <param name="ConditionRef">Upstream condition NodeId, rendered by <see cref="AlarmIndexKey"/>.</param>
/// <param name="EventId">EventId of the most recent event seen for that condition.</param>
internal readonly record struct AlarmAckTarget(string ConditionRef, byte[] EventId);
/// <summary>#619 — the key both sides of the alarm index use: server-stable (<c>nsu=</c>) when a map is
/// available, so a remote namespace-table reorder across an in-place reconnect cannot orphan it.</summary>
/// <param name="map">The current session's namespace map, or null.</param>
/// <param name="nodeId">The NodeId to key.</param>
/// <returns>The index key.</returns>
internal static string AlarmIndexKey(NamespaceMap? map, NodeId nodeId) =>
map?.ToStableReference(nodeId) ?? nodeId.ToString() ?? string.Empty;
/// <summary> /// <summary>
/// Map an OPC UA <c>BaseEventType.Severity</c> (1..1000) to our coarse-grained /// Map an OPC UA <c>BaseEventType.Severity</c> (1..1000) to our coarse-grained
/// <see cref="AlarmSeverity"/> bucket. Thresholds match the OPC UA A&amp;C Part 9 /// <see cref="AlarmSeverity"/> bucket. Thresholds match the OPC UA A&amp;C Part 9
@@ -2364,6 +2757,10 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
TransitionTo(HostState.Running); TransitionTo(HostState.Running);
_health = new DriverHealth(DriverState.Healthy, DateTime.UtcNow, null); _health = new DriverHealth(DriverState.Healthy, DateTime.UtcNow, null);
_diag.Record(OpcUaClientDiagnostics.ReconnectSuccess); _diag.Record(OpcUaClientDiagnostics.ReconnectSuccess);
// #619: transitions during the outage were missed (a recreated subscription starts empty; a transferred
// one may have overflowed). ConditionRefresh re-reports current states; unchanged ones dedup to nothing.
if (!_alarmSubscriptions.IsEmpty)
_ = Task.Run(() => RefreshAllAlarmSubscriptionsAsync(CancellationToken.None));
return; return;
} }
@@ -1,3 +1,4 @@
using Opc.Ua;
using Shouldly; using Shouldly;
using Xunit; using Xunit;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions; using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
@@ -71,6 +72,349 @@ public sealed class OpcUaClientAlarmTests
await drv.AcknowledgeAsync([], TestContext.Current.CancellationToken); await drv.AcknowledgeAsync([], TestContext.Current.CancellationToken);
} }
// ---- #619: v3 RawPath <-> upstream-source translation on the native-alarm path ----
private static VariantCollection EventFields(string sourceNode, string conditionNode) =>
[
new Variant(new byte[] { 1 }), // EventId
new Variant(new NodeId("TripAlarmType", 3)), // EventType
new Variant(NodeId.Parse(sourceNode)), // SourceNode
new Variant(new LocalizedText("tripped")), // Message
new Variant((ushort)700), // Severity
new Variant(DateTime.UtcNow), // Time
new Variant(NodeId.Parse(conditionNode)), // ConditionId
];
/// <summary>The host hands SubscribeAlarmsAsync RawPaths; the index must be keyed by the UPSTREAM source id.</summary>
[Fact]
public void BuildAlarmSourceIndex_keys_the_upstream_source_id_to_the_RawPath()
{
var index = OpcUaClientDriver.BuildAlarmSourceIndex(
["Plant/opc/plc/Trip1"],
raw => raw == "Plant/opc/plc/Trip1" ? "ns=3;s=Alarms.Trip1" : null);
index["ns=3;s=Alarms.Trip1"].ShouldBe("Plant/opc/plc/Trip1");
}
/// <summary>A ref the authored table cannot resolve is kept verbatim (direct upstream-id use stays valid).</summary>
[Fact]
public void BuildAlarmSourceIndex_keeps_an_unresolvable_ref_as_its_own_key()
{
var index = OpcUaClientDriver.BuildAlarmSourceIndex(["ns=3;s=Direct"], _ => null);
index["ns=3;s=Direct"].ShouldBe("ns=3;s=Direct");
}
/// <summary>The emitted ConditionId must be the RawPath - DriverHostActor routes on (driver, RawPath).</summary>
[Fact]
public void ProjectEvent_reports_the_RawPath_as_ConditionId_and_remembers_the_condition_node()
{
var index = new Dictionary<string, string> { ["ns=3;s=Alarms.Trip1"] = "Plant/opc/plc/Trip1" };
string? rememberedRaw = null, rememberedCond = null;
var e = OpcUaClientDriver.ProjectEvent(
EventFields("ns=3;s=Alarms.Trip1", "ns=3;s=Alarms.Trip1.Cond"), index, new FakeAlarmHandle(),
n => n.ToString(), (raw, target) => { rememberedRaw = raw; rememberedCond = target.ConditionRef; });
e.ShouldNotBeNull();
e.ConditionId.ShouldBe("Plant/opc/plc/Trip1");
e.SourceNodeId.ShouldBe("Plant/opc/plc/Trip1");
rememberedRaw.ShouldBe("Plant/opc/plc/Trip1");
rememberedCond.ShouldBe("ns=3;s=Alarms.Trip1.Cond");
}
/// <summary>An event from a source nobody authored is filtered out.</summary>
[Fact]
public void ProjectEvent_drops_an_unauthored_source()
{
var index = new Dictionary<string, string> { ["ns=3;s=Alarms.Trip1"] = "Plant/opc/plc/Trip1" };
OpcUaClientDriver.ProjectEvent(
EventFields("ns=3;s=Other", "ns=3;s=Other.Cond"), index, new FakeAlarmHandle(), n => n.ToString(), (_, _) => { })
.ShouldBeNull();
}
/// <summary>
/// A remote namespace-table reorder across an in-place SDK reconnect must not orphan the alarm index:
/// it is built once at subscribe time, while events keep arriving on the (transferred) subscription with
/// the NEW namespace indexes. Keys are therefore server-stable (<c>nsu=</c>) on both sides.
/// </summary>
[Fact]
public void ProjectEvent_matches_across_a_namespace_reorder()
{
var before = new NamespaceTable(); before.Append("urn:other"); before.Append("urn:plc"); // urn:plc = 2
var after = new NamespaceTable(); after.Append("urn:new"); after.Append("urn:other"); after.Append("urn:plc"); // urn:plc = 3
var mapBefore = NamespaceMap.FromTable(before);
var mapAfter = NamespaceMap.FromTable(after);
var index = OpcUaClientDriver.BuildAlarmSourceIndex(
["Plant/opc/plc/Trip1"], _ => OpcUaClientDriver.AlarmIndexKey(mapBefore, new NodeId("Alarms.Trip1", 2)));
var e = OpcUaClientDriver.ProjectEvent(
EventFields("ns=3;s=Alarms.Trip1", "ns=3;s=Alarms.Trip1.Cond"), index, new FakeAlarmHandle(),
n => OpcUaClientDriver.AlarmIndexKey(mapAfter, n), (_, _) => { });
e.ShouldNotBeNull();
e.ConditionId.ShouldBe("Plant/opc/plc/Trip1");
}
/// <summary>
/// A source with several conditions (Hi + HiHi on one variable) can only be told apart by the CONDITION
/// node, so a tag authored on the condition node matches on the event's ConditionId — 1:1, no collapse.
/// </summary>
[Fact]
public void ProjectEvent_matches_a_tag_authored_on_the_condition_node()
{
var index = new Dictionary<string, string>
{
["ns=3;s=Tank.HiHi"] = "Plant/opc/plc/HiHi",
["ns=3;s=Tank.Hi"] = "Plant/opc/plc/Hi",
};
var e = OpcUaClientDriver.ProjectEvent(
EventFields("ns=3;s=Tank", "ns=3;s=Tank.HiHi"), index, new FakeAlarmHandle(), n => n.ToString(), (_, _) => { });
e.ShouldNotBeNull();
e.ConditionId.ShouldBe("Plant/opc/plc/HiHi");
}
/// <summary>
/// The ack target pairs the condition node WITH the EventId of the event that named it, so Acknowledge
/// carries the real EventId (Part 9 §5.7.3) instead of an empty one a strict server rejects
/// <c>BadEventIdUnknown</c>, and the two can never come from different conditions.
/// </summary>
[Fact]
public void ProjectEvent_remembers_the_condition_with_its_EventId()
{
var index = new Dictionary<string, string> { ["ns=3;s=Alarms.Trip1"] = "Plant/opc/plc/Trip1" };
OpcUaClientDriver.AlarmAckTarget? target = null;
OpcUaClientDriver.ProjectEvent(
EventFields("ns=3;s=Alarms.Trip1", "ns=3;s=Alarms.Trip1.Cond"), index, new FakeAlarmHandle(),
n => n.ToString(), (_, t) => target = t);
target.ShouldNotBeNull();
target.Value.ConditionRef.ShouldBe("ns=3;s=Alarms.Trip1.Cond");
target.Value.EventId.ShouldBe(new byte[] { 1 });
}
// ---- #619 review: the host's NativeAlarmProjector derives Active/Acked ONLY from AlarmEventArgs.Kind; an
// Unspecified kind keeps the prior state (inactive+acked when fresh), so without a Kind no upstream alarm
// could ever show active. ----
/// <summary>ActiveState/Id and AckedState/Id are read from the two trailing select clauses.</summary>
[Fact]
public void ProjectEvent_reads_the_condition_state()
{
var index = new Dictionary<string, string> { ["ns=3;s=Alarms.Trip1"] = "Plant/opc/plc/Trip1" };
var fields = EventFields("ns=3;s=Alarms.Trip1", "ns=3;s=Alarms.Trip1.Cond");
fields.Add(new Variant(true)); // ActiveState/Id
fields.Add(new Variant(false)); // AckedState/Id
OpcUaClientDriver.ProjectEvent(fields, index, new FakeAlarmHandle(), n => n.ToString(), (_, _) => { },
out var state);
state.ShouldBe((true, false));
}
/// <summary>The transition sequence drives a projector from the driver's last-told state to the event's.</summary>
[Theory]
[InlineData(false, true, true, false, new[] { AlarmTransitionKind.Raise })]
[InlineData(false, true, true, true, new[] { AlarmTransitionKind.Raise, AlarmTransitionKind.Acknowledge })]
[InlineData(true, false, true, true, new[] { AlarmTransitionKind.Acknowledge })]
[InlineData(true, false, false, false, new[] { AlarmTransitionKind.Clear })]
[InlineData(true, true, false, true, new[] { AlarmTransitionKind.Clear })]
[InlineData(false, false, false, true, new[] { AlarmTransitionKind.Acknowledge })]
[InlineData(true, false, false, true, new[] { AlarmTransitionKind.Clear, AlarmTransitionKind.Acknowledge })]
[InlineData(true, true, true, false, new[] { AlarmTransitionKind.Retrigger })]
[InlineData(true, false, true, false, new AlarmTransitionKind[0])]
public void AlarmTransitions_move_prior_to_target(
bool priorActive, bool priorAcked, bool active, bool acked, AlarmTransitionKind[] expected)
{
OpcUaClientDriver.AlarmTransitions((priorActive, priorAcked), (active, acked)).ShouldBe(expected);
}
// ---- #619 re-review: the transition table must reach EVERY target from EVERY prior. ----
/// <summary>The host projector's semantics, restated: the only thing the driver's kinds act upon.</summary>
private static (bool Active, bool Acked) Simulate((bool Active, bool Acked) s, AlarmTransitionKind k) => k switch
{
AlarmTransitionKind.Raise or AlarmTransitionKind.Retrigger => (true, false),
AlarmTransitionKind.Acknowledge => (s.Active, true),
AlarmTransitionKind.Clear => (false, s.Acked),
_ => s,
};
/// <summary>
/// Exhaustive over all 16 (prior, target) pairs: replaying the emitted kinds through the projector's
/// semantics lands on the target, and an unchanged state emits nothing. The table this replaced reached
/// (inactive, acked) from (active, acked) → (inactive, UNACKED) wrongly, and could not express a
/// cleared-but-unacked condition first seen by a ConditionRefresh at all.
/// </summary>
[Fact]
public void AlarmTransitions_reach_every_target_from_every_prior()
{
bool[] b = [false, true];
foreach (var pa in b) foreach (var pk in b) foreach (var ta in b) foreach (var tk in b)
{
var kinds = OpcUaClientDriver.AlarmTransitions((pa, pk), (ta, tk));
var s = (pa, pk);
foreach (var k in kinds) s = Simulate(s, k);
s.ShouldBe((ta, tk), $"prior ({pa},{pk}) -> target ({ta},{tk}) via [{string.Join(",", kinds)}]");
if ((pa, pk) == (ta, tk)) kinds.ShouldBeEmpty();
kinds.ShouldNotContain(AlarmTransitionKind.Unspecified);
}
}
private static VariantCollection ConditionFields(
bool active, bool acked, string message = "tripped", ushort severity = 700, byte[]? eventId = null,
NodeId? branchId = null, NodeId? eventType = null)
{
var f = EventFields("ns=3;s=Alarms.Trip1", "ns=3;s=Alarms.Trip1.Cond");
f[0] = new Variant(eventId ?? new byte[] { 1 });
if (eventType is not null) f[1] = new Variant(eventType);
f[3] = new Variant(new LocalizedText(message));
f[4] = new Variant(severity);
f.Add(new Variant(active));
f.Add(new Variant(acked));
f.Add(branchId is null ? Variant.Null : new Variant(branchId));
return f;
}
private static readonly Dictionary<string, string> TripIndex = new() { ["ns=3;s=Alarms.Trip1"] = "Plant/opc/plc/Trip1" };
private static List<AlarmEventArgs> Record(OpcUaClientDriver drv)
{
var seen = new List<AlarmEventArgs>();
drv.OnAlarmEvent += (_, e) => { lock (seen) seen.Add(e); };
return seen;
}
/// <summary>The remembered EventId is a COPY — the decoded notification's array is not ours to alias.</summary>
[Fact]
public void ProjectEvent_copies_the_EventId()
{
var raw = new byte[] { 7, 8, 9 };
OpcUaClientDriver.AlarmAckTarget? target = null;
OpcUaClientDriver.ProjectEvent(ConditionFields(true, false, eventId: raw), TripIndex, new FakeAlarmHandle(),
n => n.ToString(), (_, t) => target = t);
raw[0] = 0;
target!.Value.EventId.ShouldBe(new byte[] { 7, 8, 9 });
}
/// <summary>
/// A condition BRANCH event (BranchId non-null) reports a previous state of the condition and carries its
/// own EventId — projecting it would flip the trunk's state, and remembering it would ack the branch.
/// </summary>
[Fact]
public void ProjectEvent_ignores_a_condition_branch_event()
{
var remembered = false;
OpcUaClientDriver.ProjectEvent(
ConditionFields(false, false, branchId: new NodeId(Guid.NewGuid(), 3)), TripIndex, new FakeAlarmHandle(),
n => n.ToString(), (_, _) => remembered = true)
.ShouldBeNull();
remembered.ShouldBeFalse();
// Positive control: the same event with a null BranchId (the trunk) projects.
OpcUaClientDriver.ProjectEvent(ConditionFields(false, false), TripIndex, new FakeAlarmHandle(),
n => n.ToString(), (_, _) => { })
.ShouldNotBeNull();
}
/// <summary>RefreshStart/RefreshEnd are markers around a ConditionRefresh, never alarms — even pass-through.</summary>
[Fact]
public void Refresh_markers_are_never_projected_as_alarms()
{
foreach (var type in new[] { ObjectTypeIds.RefreshStartEventType, ObjectTypeIds.RefreshEndEventType, ObjectTypeIds.RefreshRequiredEventType })
{
var fields = EventFields("i=2253", "i=0");
fields[1] = new Variant(type);
OpcUaClientDriver.ProjectEvent(fields, new Dictionary<string, string>(), new FakeAlarmHandle(),
n => n.ToString(), (_, _) => { }).ShouldBeNull($"{type} is a refresh marker");
}
OpcUaClientDriver.IsRefreshRequired(EventFieldsOfType(ObjectTypeIds.RefreshRequiredEventType)).ShouldBeTrue();
OpcUaClientDriver.IsRefreshRequired(EventFieldsOfType(ObjectTypeIds.RefreshEndEventType)).ShouldBeFalse();
}
private static VariantCollection EventFieldsOfType(NodeId type)
{
var f = EventFields("i=2253", "i=0");
f[1] = new Variant(type);
return f;
}
/// <summary>
/// A ConditionRefresh re-delivers every retained condition's CURRENT state. One the host was already told
/// must emit nothing — no re-raise (which would un-ack an acked alarm) and no Unspecified (which a host
/// projector reset by a redeploy would project from its inactive default).
/// </summary>
[Fact]
public void A_refresh_of_an_already_reported_state_emits_nothing()
{
using var drv = new OpcUaClientDriver(new OpcUaClientDriverOptions(), "opcua-alarm-dedup");
var seen = Record(drv);
var handle = new FakeAlarmHandle();
drv.RouteAlarmEvent(ConditionFields(true, false), TripIndex, handle);
drv.RouteAlarmEvent(ConditionFields(true, true), TripIndex, handle);
seen.Select(e => e.Kind).ShouldBe([AlarmTransitionKind.Raise, AlarmTransitionKind.Acknowledge]);
drv.RouteAlarmEvent(ConditionFields(true, true), TripIndex, handle); // the refresh copy
seen.Count.ShouldBe(2);
// Positive control: a genuine message change on an unchanged state still reaches the host.
drv.RouteAlarmEvent(ConditionFields(true, true, message: "worse"), TripIndex, handle);
seen.Count.ShouldBe(3);
seen[2].Kind.ShouldBe(AlarmTransitionKind.Unspecified);
seen[2].Message.ShouldBe("worse");
}
/// <summary>The exact upstream severity rides OpcUaSeverity (#626) so the host does not re-bucket it.</summary>
[Fact]
public void The_exact_upstream_severity_is_carried()
{
using var drv = new OpcUaClientDriver(new OpcUaClientDriverOptions(), "opcua-alarm-sev");
var seen = Record(drv);
drv.RouteAlarmEvent(ConditionFields(true, false, severity: 650), TripIndex, new FakeAlarmHandle());
seen.Single().OpcUaSeverity.ShouldBe(650);
}
/// <summary>
/// A driver re-initialise (comms-loss watchdog, Force Reconnect) re-subscribes the SAME set over nodes the
/// host did not rebuild, so the last-told state must survive it — or the refresh after it would re-raise
/// every acked active alarm (un-ack). A CHANGED set means the host rebuilt (nodes recreated inactive), so
/// there the state is forgotten and the refresh re-raises from scratch.
/// </summary>
[Fact]
public async Task Last_told_state_survives_a_reinit_of_the_same_set_but_not_a_set_change()
{
using var drv = new OpcUaClientDriver(new OpcUaClientDriverOptions(), "opcua-alarm-reinit");
var seen = Record(drv);
var handle = new FakeAlarmHandle();
drv.BeginAlarmSubscription(["Plant/opc/plc/Trip1"]);
drv.RouteAlarmEvent(ConditionFields(true, true), TripIndex, handle);
seen.Count.ShouldBe(2);
await drv.ShutdownAsync(TestContext.Current.CancellationToken);
drv.BeginAlarmSubscription(["Plant/opc/plc/Trip1"]);
drv.RouteAlarmEvent(ConditionFields(true, true), TripIndex, handle);
seen.Count.ShouldBe(2, "same set after a reinit: the refresh copy is a duplicate");
drv.BeginAlarmSubscription(["Plant/opc/plc/Trip1", "Plant/opc/plc/Trip2"]);
drv.RouteAlarmEvent(ConditionFields(true, true), TripIndex, handle);
seen.Skip(2).Select(e => e.Kind).ShouldBe([AlarmTransitionKind.Raise, AlarmTransitionKind.Acknowledge]);
}
/// <summary>A server without ConditionRefresh is tolerated; anything else is a real failure.</summary>
[Fact]
public void ConditionRefresh_unsupported_status_codes()
{
OpcUaClientDriver.IsConditionRefreshUnsupported(StatusCodes.BadNotSupported).ShouldBeTrue();
OpcUaClientDriver.IsConditionRefreshUnsupported(StatusCodes.BadMethodInvalid).ShouldBeTrue();
OpcUaClientDriver.IsConditionRefreshUnsupported(StatusCodes.BadNotImplemented).ShouldBeTrue();
OpcUaClientDriver.IsConditionRefreshUnsupported(StatusCodes.BadTimeout).ShouldBeFalse();
}
private sealed class FakeAlarmHandle : IAlarmSubscriptionHandle private sealed class FakeAlarmHandle : IAlarmSubscriptionHandle
{ {
/// <summary>Gets the diagnostic identifier for this alarm handle.</summary> /// <summary>Gets the diagnostic identifier for this alarm handle.</summary>
@@ -255,6 +255,68 @@ public sealed class OpcUaClientDriverWireTests
"the writable variable and the read-only one must classify differently"); "the writable variable and the read-only one must classify differently");
} }
/// <summary>
/// #619 — ConditionRefresh over the wire: a condition already active+unacked BEFORE the driver subscribes is
/// reported (Raise, keyed by the RawPath) right after subscribe, not only on its next transition. A second
/// subscribe of the same set refreshes again and must emit nothing (no re-raise); a positive control then
/// proves the feed is still live by acknowledging upstream and seeing exactly the Acknowledge arrive.
/// </summary>
[Fact]
public async Task A_condition_already_active_before_subscribe_is_reported_by_ConditionRefresh()
{
const string AlarmRef = "Sim/dev1/TempHi";
var ct = TestContext.Current.CancellationToken;
await using var rig = await StartRigAsync("drv-alarm-refresh", ct);
rig.MaterialiseNative();
rig.Project(native: true, new AlarmConditionSnapshot(Active: true, Acknowledged: false, Confirmed: true,
Enabled: true, AlarmShelvingKind.Unshelved, 800, "already hot"));
await using var drv = new OpcUaClientDriver(new OpcUaClientDriverOptions
{
EndpointUrl = rig.EndpointUrl,
RawTags = [Tag(AlarmRef, ServerEventRig.RawNsu(ServerEventRig.RawAlarmPath))],
SecurityPolicy = OpcUaSecurityPolicy.None,
SecurityMode = OpcUaSecurityMode.None,
AuthType = OpcUaAuthType.Username,
Username = "apikey",
Password = ServerEventRig.KeyPassword,
AutoAcceptCertificates = true,
TargetNamespaceKind = OpcUaTargetNamespaceKind.SystemPlatform,
Timeout = TimeSpan.FromSeconds(10),
SessionTimeout = TimeSpan.FromSeconds(30),
KeepAliveInterval = TimeSpan.FromSeconds(1),
}, "wire-alarm-refresh");
await drv.InitializeAsync("{}", ct);
var seen = new List<AlarmEventArgs>();
drv.OnAlarmEvent += (_, e) => { lock (seen) seen.Add(e); };
AlarmEventArgs[] Seen() { lock (seen) return [.. seen]; }
var first = await drv.SubscribeAlarmsAsync([AlarmRef], ct);
await RecordedEvents.WaitUntilAsync(() => Seen().Any(e => e.Kind == AlarmTransitionKind.Raise));
var raise = Seen().Single(e => e.Kind == AlarmTransitionKind.Raise);
raise.ConditionId.ShouldBe(AlarmRef);
raise.Message.ShouldBe("already hot");
raise.OpcUaSeverity.ShouldBe(800);
Seen().ShouldAllBe(e => e.ConditionId == AlarmRef, "refresh markers must never surface as alarms");
// Same set again, released first exactly as DriverInstanceActor does → a fresh subscription and another
// ConditionRefresh; the identical state must not re-raise.
var before = Seen().Length;
await drv.UnsubscribeAlarmsAsync(first, ct);
await drv.SubscribeAlarmsAsync([AlarmRef], ct);
await Task.Delay(1500, ct); // let the second refresh's events land before the control transition
// Positive control: an upstream acknowledge arrives as exactly one Acknowledge, after which nothing else.
rig.Project(native: true, new AlarmConditionSnapshot(Active: true, Acknowledged: true, Confirmed: true,
Enabled: true, AlarmShelvingKind.Unshelved, 800, "already hot"));
await RecordedEvents.WaitUntilAsync(() => Seen().Any(e => e.Kind == AlarmTransitionKind.Acknowledge));
await Task.Delay(500, ct);
Seen().Skip(before).Select(e => e.Kind).ShouldBe([AlarmTransitionKind.Acknowledge],
"the second refresh re-delivered an unchanged state and must have emitted nothing");
}
private sealed class RecordingBuilder : IAddressSpaceBuilder private sealed class RecordingBuilder : IAddressSpaceBuilder
{ {
public List<(string BrowseName, DriverAttributeInfo Info)> Variables { get; } = []; public List<(string BrowseName, DriverAttributeInfo Info)> Variables { get; } = [];