Merge #838 review fixes: per-condition deferral with liveness checks, lockout identity canonicalisation
This commit is contained in:
Joseph Doherty
@@ -208,7 +208,11 @@ Two things to know before relying on any of this:
|
||||
notifier's `MonitoredNode2` drains a BOUNDED channel (4096, `FullMode = Wait`) and its delivery task takes `Lock` on a
|
||||
verdict-cache miss (SDK `GetNodeMetadata` — flag off too — and #826's check); sync `ReportEvent` blocks on a full
|
||||
channel ⇒ a >4096 burst under one `Lock` hold deadlocked the node manager (stalled at 4097). Condition events go
|
||||
through `ReportEventWithoutBlocking` (async on overflow, ordered chain). Folder events follow Part 3's shape —
|
||||
through `ReportEventWithoutBlocking`: our own copy of the SDK walk with a LIVENESS check before every hop (and in
|
||||
`OnConditionReportEvent`) — a deferred event of a removed/replaced condition is dropped, never a ghost on `i=2253`;
|
||||
per-CONDITION ordered queues on overflow; coalesce past 1024/condition or 65 536 total (oldest not-started gives way
|
||||
to newest); Warning/Information + `otopcua.opcua.condition_events.{deferred,discarded}`. ConditionRefresh does not
|
||||
flush the queues (documented residual). Folder events follow Part 3's shape —
|
||||
`ReceiveEvents` on EventType + SourceNode (SDK check, kept first) AND our decision; non-condition events on a
|
||||
folder item fail closed under enforcement. Tests: `FolderEventFloodWireTests`.
|
||||
- **ns=0 standard nodes stay ungated by design, even with `Security:Acl:Enabled`** (#617) — `ServerStatus`,
|
||||
@@ -252,7 +256,8 @@ Two things to know before relying on any of this:
|
||||
for parallelism — ScadaBridge opens many keyed sessions from one IP), refused only on lockout or, uncounted, after
|
||||
`SlotWaitSeconds`. An expired lockout resets the count. On
|
||||
`None` endpoints the key is effectively IP-only (NAT neighbours share it, UserName only). ⚠️ **Two counts (#844):**
|
||||
per (client, credential identity = `IOpcUaUserAuthenticator.DescribeCredential`, `key:<keyId>` / `user:<name>`),
|
||||
per (client, CANONICAL credential identity = `IOpcUaUserAuthenticator.DescribeCredential`, `key:<keyId>` trimmed +
|
||||
lower-cased, non-hex ids folded together, since SQL `=` is CI/trailing-space-blind / `user:<name>`),
|
||||
cleared by a success with THAT identity, locks that identity; per client (`MaxFailedAttemptsPerClient`, default
|
||||
2 × max) — **never reset by a success** (else a valid key launders guesses at another key id) — locks every
|
||||
credential. ⚠️ **Only definite bad credentials count (#845):** store timeout / miss-budget / store throw / unset
|
||||
|
||||
+40
-8
@@ -335,7 +335,11 @@ The host now sets `MaxFailedAuthenticationAttempts = 0` on every configuration i
|
||||
- **Two counts per client (#844).** Each *definite* failure is charged to the (client, **credential identity**)
|
||||
pair and to the client as a whole. The credential identity is `IOpcUaUserAuthenticator.DescribeCredential` —
|
||||
for API keys `key:<keyId>` (the public lookup index, never the secret), for anything that is not a well-formed key
|
||||
`user:<username>`.
|
||||
`user:<username>`. It is **canonical**: the ConfigDb store matches the key id with SQL Server `=` (case-insensitive,
|
||||
trailing spaces ignored), so the id is trimmed and lower-cased, and an id outside the minted lowercase-hex charset
|
||||
(which no real key carries) folds into one shared `key:<not-a-minted-id>` identity; the username is trimmed and
|
||||
lower-cased too. Otherwise `otua_abc…`, `otua_ABC…` and `otua_abc… ` would each get a fresh per-credential count for
|
||||
one key.
|
||||
- **Per credential:** at `MaxFailedAttempts` (5) failures of one identity, each within `FailureWindowSeconds` (60) of
|
||||
the previous, that client is refused **that identity** `BadUserAccessDenied` for `LockoutSeconds` (300), without
|
||||
the key store being consulted. A success with **that same identity** clears its count.
|
||||
@@ -852,13 +856,41 @@ Behaviours worth knowing before enabling it:
|
||||
own `GetNodeMetadata` (so **with the flag off too**, i.e. this predates #826) and in the #826 check. A burst of more
|
||||
than 4096 events under one hold of `Lock`, starting at a miss (a new item, or right after a deploy invalidated the
|
||||
verdicts), therefore deadlocked the node manager: `FolderEventFloodWireTests` stalled at exactly 4097 events, flag on
|
||||
and off. Condition events now go through `ReportEventWithoutBlocking`: `ReportEventAsync` completes inline while the
|
||||
channels have room (the normal case, unchanged); when one is full that report continues asynchronously, holding no
|
||||
thread and no lock, and later reports queue behind it so events still leave **in report order**. The deferred work
|
||||
holds only frozen snapshots — bounded by the burst — and drains once `Lock` is released; what a client receives is
|
||||
still bounded by its item's queue (oldest discarded). Pinned by `FolderEventFloodWireTests` (6000 events under one
|
||||
`Lock` hold with publishing disabled: completes, deferred reports drain to zero, the item delivers at most its
|
||||
revised `QueueSize` in increasing order ending with the last event, and the next event is delivered).
|
||||
and off. Condition events now go through `ReportEventWithoutBlocking`:
|
||||
- **Delivery** walks the same path as the SDK's `ReportEventAsync` (the condition's sync `OnReportEvent` → Server-object
|
||||
items, its own `MonitoredNode2`, then each inverse notifier recursively from a snapshot of the notifier list), and
|
||||
completes **inline** while the channels have room — the normal case, unchanged. It re-checks **before every hop** that
|
||||
the condition is still the live instance, and `OnConditionReportEvent` checks the same, so an event that waited past a
|
||||
deploy that removed or replaced its condition is **dropped, never delivered as a ghost** (a ghost Retain alarm after
|
||||
the NodeDeleted for ScadaBridge on `i=2253`; with the ACL on, a non-live scripted condition could otherwise be admitted
|
||||
through its OLD equipment folder, #808).
|
||||
- **Deferral is per condition.** When a channel is full that event continues asynchronously (no thread, no lock held)
|
||||
and the condition gets a queue; its later events wait behind it, so **one condition's events stay in report order**,
|
||||
while conditions whose channels have room are not held up. Events of different conditions may interleave differently
|
||||
from report order — each carries its condition's full state, so nothing a client derives depends on that.
|
||||
- **Overflow policy — coalesce, newest wins.** A condition holds at most 1024 waiting events; past that, or while more
|
||||
than 65 536 are deferred in total, its **oldest not-yet-started** event is discarded. A condition event carries the
|
||||
condition's complete state, so for every Retain-based consumer the newest supersedes the rest; what is lost is
|
||||
intermediate transitions on the OPC UA event stream only (alarm history is recorded on the engine/driver path, not
|
||||
from these events). Dropping the newest would leave subscribers on a stale state; blocking is the deadlock above. Each
|
||||
condition always keeps its newest, so the ceiling is 65 536 + the number of conditions.
|
||||
- **Visibility:** entering deferral logs a **Warning** (at most once a minute) naming the first condition; draining logs
|
||||
an **Information** line with the episode's deferred/discarded counts. Meters: `otopcua.opcua.condition_events.deferred`
|
||||
(live depth, up/down) and `otopcua.opcua.condition_events.discarded{reason=coalesced|not_live}`.
|
||||
- **SDK behaviour verified (2.0.1, decompiled):** `NodeState.ReportEventAsync`/`GetNotifiers` copy the notifier list under
|
||||
the node's own notifier lock, so a concurrent `RemoveNotifier` under `Lock` cannot break a deferred walk (it only
|
||||
changes which folders a later hop reaches — and the liveness check covers removal); `MonitoredNode2.Dispose` completes
|
||||
its channel writer, which fails a pending `WriteAsync` with `ChannelClosedException` (swallowed by the SDK), so a
|
||||
disposed node cannot stall a condition's queue; and the delivery task catches per-event exceptions, so it keeps draining.
|
||||
- ⚠️ **Residual (documented, not fixed): `ConditionRefresh` does not flush deferrals.** While a notifier channel is
|
||||
overloaded, a client that refreshes may receive older deferred events of a condition after `RefreshEndEvent`. They
|
||||
are still in that condition's order and its newest event arrives last, so the state a client ends up holding is
|
||||
correct; making refresh wait for the queues would block a service call on the very drain the overload is stalling.
|
||||
|
||||
Pinned by `FolderEventFloodWireTests`: 6000 events under one `Lock` hold with publishing disabled (flag on and off)
|
||||
complete, drain to zero, coalesce, and the item delivers at most its revised `QueueSize` in increasing order ending
|
||||
with the last event; and a condition **removed** in the same `Lock` hold has none of its deferred events delivered to
|
||||
either the folder or the Server-object item (the flag-off variant fails without the liveness checks).
|
||||
|
||||
**Model-change and audit events on the Server object are gated too** (Gitea #588). The node manager's
|
||||
own server-wide events — `GeneralModelChangeEvent` (NodeAdded / NodeDeleted / DataTypeChanged) and the
|
||||
|
||||
@@ -154,6 +154,24 @@ public static class OtOpcUaTelemetry
|
||||
Meter.CreateCounter<long>("otopcua.opcua.apikey_store.bounded", unit: "{lookup}",
|
||||
description: "API-key store lookups the bound intervened on (outcome=timeout|cache_hit).");
|
||||
|
||||
/// <summary>
|
||||
/// Part 9 condition events whose delivery is currently deferred because an event notifier's SDK delivery
|
||||
/// channel was full (Gitea #838). Zero in steady state; non-zero means some folder's event subscribers are
|
||||
/// not keeping up with the alarm event rate (or the node manager lock is held across a large burst).
|
||||
/// </summary>
|
||||
public static readonly UpDownCounter<long> ConditionEventReportsDeferred =
|
||||
Meter.CreateUpDownCounter<long>("otopcua.opcua.condition_events.deferred", unit: "{event}",
|
||||
description: "Condition events waiting for room in an event notifier's delivery channel.");
|
||||
|
||||
/// <summary>
|
||||
/// Deferred condition events discarded before delivery (Gitea #838): <c>reason=coalesced</c> — the per-condition
|
||||
/// deferral budget was exceeded and the condition's oldest not-yet-started event gave way to its newest;
|
||||
/// <c>reason=not_live</c> — the condition was removed or replaced before its deferred event was delivered.
|
||||
/// </summary>
|
||||
public static readonly Counter<long> ConditionEventReportsDiscarded =
|
||||
Meter.CreateCounter<long>("otopcua.opcua.condition_events.discarded", unit: "{event}",
|
||||
description: "Deferred condition events discarded before delivery (reason=coalesced|not_live).");
|
||||
|
||||
// ---------------- Convenience helpers ----------------
|
||||
|
||||
/// <summary>
|
||||
|
||||
@@ -168,8 +168,23 @@ public sealed class ApiKeyOpcUaUserAuthenticator : IOpcUaUserAuthenticator
|
||||
// lmxopcua#844 — the key id is public (it is the lookup index, never the secret), so guesses at different keys
|
||||
// are counted apart and a client's own successful key cannot clear the count of a key it is guessing at.
|
||||
// A credential that is not a well-formed key falls back to the username (all such guesses share one count).
|
||||
public string DescribeCredential(string username, string password) =>
|
||||
ApiKeySecret.TryParse(password, out var apiKeyId, out _) ? "key:" + apiKeyId : "user:" + username;
|
||||
//
|
||||
// #844 review: the identity is CANONICAL, never the raw presented id. ConfigDbApiKeyStore matches the id with SQL
|
||||
// Server `=` under a case-insensitive collation that also ignores trailing spaces, so `otua_abc…`, `otua_ABC…` and
|
||||
// `otua_abc… ` reach the same key row — as separate lockout identities they would each get a fresh per-credential
|
||||
// count. Minted ids are lowercase hex (ApiKeySecret.Generate), so: trim, lowercase, and fold every id outside that
|
||||
// charset (which no real key can carry) into one shared identity. The username is folded the same way. Merging
|
||||
// only ever makes the per-credential count stricter; the per-client cap is unaffected.
|
||||
public string DescribeCredential(string username, string password)
|
||||
{
|
||||
if (!ApiKeySecret.TryParse(password, out var apiKeyId, out _))
|
||||
return "user:" + (username ?? string.Empty).Trim().ToLowerInvariant();
|
||||
|
||||
var canonical = apiKeyId.Trim().ToLowerInvariant();
|
||||
return canonical.Length > 0 && canonical.All(char.IsAsciiHexDigitLower)
|
||||
? "key:" + canonical
|
||||
: "key:<not-a-minted-id>";
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
// Resolves the <c>anonymous</c> principal's grants through the same mapper a key's principal
|
||||
|
||||
@@ -4,6 +4,7 @@ using Microsoft.Extensions.Logging;
|
||||
using Microsoft.Extensions.Logging.Abstractions;
|
||||
using Opc.Ua;
|
||||
using Opc.Ua.Server;
|
||||
using ZB.MOM.WW.OtOpcUa.Commons.Observability;
|
||||
using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
|
||||
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
||||
using ZB.MOM.WW.OtOpcUa.Core.Authorization;
|
||||
@@ -1400,22 +1401,44 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>Serializes <see cref="_eventReportChain"/>.</summary>
|
||||
private readonly object _eventReportGate = new();
|
||||
|
||||
/// <summary>The tail of the deferred condition-event reports (Gitea #838); completed when none is pending.</summary>
|
||||
private Task _eventReportChain = Task.CompletedTask;
|
||||
|
||||
private int _deferredEventReports;
|
||||
|
||||
/// <summary>Condition-event reports currently deferred because a notifier's delivery channel was full (test observation).</summary>
|
||||
internal int DeferredEventReports => Volatile.Read(ref _deferredEventReports);
|
||||
/// <summary>The most deferred events one condition may hold before its oldest not-yet-started one gives way (#838).</summary>
|
||||
internal const int MaxDeferredEventsPerCondition = 1024;
|
||||
|
||||
/// <summary>
|
||||
/// Reports a condition event through the condition's notifier chain WITHOUT blocking the caller, in report
|
||||
/// order (Gitea #838).
|
||||
/// The total deferred events across all conditions beyond which each new deferral coalesces its condition's
|
||||
/// queue down to its newest event (#838). Each condition always keeps its newest, so the true ceiling is this
|
||||
/// plus the number of conditions.
|
||||
/// </summary>
|
||||
/// <param name="source">The condition.</param>
|
||||
internal const int MaxDeferredEventsTotal = 65_536;
|
||||
|
||||
private static readonly TimeSpan DeferralWarningInterval = TimeSpan.FromMinutes(1);
|
||||
|
||||
/// <summary>Guards <see cref="_deferredReports"/> and the deferral episode bookkeeping. Never held across a delivery.</summary>
|
||||
private readonly object _eventReportGate = new();
|
||||
|
||||
/// <summary>Conditions with deferred events, each with its own ordered queue (#838). Keyed by instance.</summary>
|
||||
private readonly Dictionary<AlarmConditionState, DeferredConditionReports> _deferredReports =
|
||||
new(System.Collections.Generic.ReferenceEqualityComparer.Instance);
|
||||
|
||||
private int _deferredEventReports;
|
||||
private long _discardedEventReports;
|
||||
private DateTime _lastDeferralWarningUtc = DateTime.MinValue;
|
||||
private bool _episodeWarned;
|
||||
private long _episodeDeferred;
|
||||
private long _episodeDiscarded;
|
||||
|
||||
/// <summary>Condition events currently deferred — queued or in flight (test observation).</summary>
|
||||
internal int DeferredEventReports => Volatile.Read(ref _deferredEventReports);
|
||||
|
||||
/// <summary>Deferred condition events discarded before delivery, any reason (test observation).</summary>
|
||||
internal long DiscardedEventReports => Interlocked.Read(ref _discardedEventReports);
|
||||
|
||||
/// <summary>
|
||||
/// Reports a condition event through the condition's notifier chain WITHOUT blocking the caller, in the order
|
||||
/// this condition's events were reported, and never for a condition that is no longer live (Gitea #838).
|
||||
/// </summary>
|
||||
/// <param name="condition">The condition (must be the live instance — every caller holds <c>Lock</c>).</param>
|
||||
/// <param name="e">The frozen snapshot of the event.</param>
|
||||
/// <remarks>
|
||||
/// <para>
|
||||
@@ -1423,72 +1446,212 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3
|
||||
/// <c>MonitoredNode2</c> whose delivery task drains a BOUNDED channel (4096, <c>FullMode = Wait</c>), and
|
||||
/// the synchronous <c>ReportEvent</c> BLOCKS when that channel is full. Condition events are reported
|
||||
/// under <c>Lock</c>, and the delivery task takes <c>Lock</c> on every per-event verdict-cache miss (the
|
||||
/// SDK's own <c>GetNodeMetadata</c>, and #826's <see cref="IsNotifierEventVisibleTo"/>). So a burst of more
|
||||
/// than 4096 events under one hold of <c>Lock</c>, starting at a miss, parked the consumer on <c>Lock</c>
|
||||
/// while the producer waited on the consumer: a deadlock that froze the whole node manager (pinned by
|
||||
/// <c>FolderEventFloodWireTests</c>, which stalled at exactly 4097 events before this).
|
||||
/// SDK's own <c>GetNodeMetadata</c> — flag off too — and #826's <see cref="IsNotifierEventVisibleTo"/>). So
|
||||
/// a burst of more than 4096 events under one hold of <c>Lock</c>, starting at a miss, deadlocked the
|
||||
/// node manager (<c>FolderEventFloodWireTests</c> stalled at exactly 4097 events before this).
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// <b>What this does.</b> The common case is unchanged and synchronous: <c>ReportEventAsync</c> completes
|
||||
/// inline while every channel has room. Only when a channel is full does that report continue
|
||||
/// asynchronously (awaiting channel space, holding no thread and no lock), and every later report queues
|
||||
/// behind it, so events still leave in the order they were reported. The deferred work holds only frozen
|
||||
/// snapshots, so it is bounded by the burst that caused it and drains once <c>Lock</c> is released.
|
||||
/// Delivery to clients stays bounded by each monitored item's own queue (oldest discarded). A deferred
|
||||
/// report runs the condition's synchronous <c>OnReportEvent</c> hook (<see cref="OnConditionReportEvent"/>)
|
||||
/// off <c>Lock</c>; that hook takes <c>Lock</c> itself where it reads guarded state.
|
||||
/// <b>Delivery.</b> <see cref="DeliverConditionEventAsync"/> walks the same path the SDK's
|
||||
/// <c>ReportEventAsync</c> does (the node's sync <c>OnReportEvent</c>, its <c>MonitoredNode2</c>, then each
|
||||
/// inverse notifier recursively, from a snapshot of the notifier list) — but re-checks before EVERY hop that
|
||||
/// the condition is still the live instance, so an event deferred past a deploy that removed or replaced
|
||||
/// it is dropped rather than delivered as a ghost (review of 91e28e87). The common case is unchanged and
|
||||
/// synchronous: the walk completes inline while every channel has room.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// <b>Deferral.</b> When a channel is full the walk continues asynchronously (holding no thread or lock),
|
||||
/// and the condition gets a queue: its later events wait behind it, so ONE condition's events stay in
|
||||
/// order, while other conditions whose channels have room are not held up (per-condition, not global).
|
||||
/// Events of different conditions may interleave differently than reported — each carries its own
|
||||
/// condition's full state, so nothing a client derives depends on cross-condition order.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// <b>Overflow policy: coalesce, newest wins.</b> A condition holds at most
|
||||
/// <see cref="MaxDeferredEventsPerCondition"/> waiting events; past that, or while the total exceeds
|
||||
/// <see cref="MaxDeferredEventsTotal"/>, its OLDEST not-yet-started event is discarded and counted
|
||||
/// (<c>otopcua.opcua.condition_events.discarded{reason=coalesced}</c>). A condition event carries the
|
||||
/// condition's complete state, so the newest supersedes the older ones for every Retain-based consumer
|
||||
/// (ScadaBridge); what is lost is intermediate transitions on the OPC UA event stream only, never the
|
||||
/// final state (alarm history is recorded by the engine/driver path, not from these events). Dropping
|
||||
/// the newest instead would leave subscribers on a stale state; blocking is the deadlock above.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// <b>Visibility.</b> Entering deferral logs a Warning (at most once a minute), leaving it an Information
|
||||
/// line with the episode's counts; <c>otopcua.opcua.condition_events.deferred</c> is the live depth.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
private void ReportEventWithoutBlocking(NodeState source, IFilterTarget e)
|
||||
private void ReportEventWithoutBlocking(AlarmConditionState condition, IFilterTarget e)
|
||||
{
|
||||
lock (_eventReportGate)
|
||||
{
|
||||
if (_eventReportChain.IsCompleted)
|
||||
if (_deferredReports.TryGetValue(condition, out var queued))
|
||||
{
|
||||
var report = source.ReportEventAsync(SystemContext, e);
|
||||
if (report.IsCompletedSuccessfully) return;
|
||||
Interlocked.Increment(ref _deferredEventReports);
|
||||
_eventReportChain = ObserveEventReportAsync(report.AsTask(), source);
|
||||
EnqueueDeferred(condition, queued, e);
|
||||
return;
|
||||
}
|
||||
|
||||
Interlocked.Increment(ref _deferredEventReports);
|
||||
_eventReportChain = ReportAfterAsync(_eventReportChain, source, e);
|
||||
}
|
||||
}
|
||||
|
||||
private async Task ReportAfterAsync(Task previous, NodeState source, IFilterTarget e)
|
||||
{
|
||||
await previous.ConfigureAwait(false); // never faults: every link observes its own failure
|
||||
Task report;
|
||||
// Every caller holds Lock, so no other report of this condition can start until this returns; the gate is
|
||||
// NOT held across the delivery (the delivery may take Lock, and gate→Lock against Lock→gate would deadlock).
|
||||
ValueTask delivery;
|
||||
try
|
||||
{
|
||||
report = source.ReportEventAsync(SystemContext, e).AsTask();
|
||||
delivery = DeliverConditionEventAsync(condition, condition, e);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
report = Task.FromException(ex);
|
||||
delivery = ValueTask.FromException(ex);
|
||||
}
|
||||
|
||||
await ObserveEventReportAsync(report, source).ConfigureAwait(false);
|
||||
if (delivery.IsCompletedSuccessfully) return;
|
||||
|
||||
lock (_eventReportGate)
|
||||
{
|
||||
var queue = new DeferredConditionReports();
|
||||
_deferredReports[condition] = queue;
|
||||
CountDeferred(+1, condition);
|
||||
_ = DrainDeferredAsync(condition, queue, delivery.AsTask());
|
||||
}
|
||||
}
|
||||
|
||||
private async Task ObserveEventReportAsync(Task report, NodeState source)
|
||||
/// <summary>Queues a later event behind a condition's deferred one, coalescing past the budget. Under the gate.</summary>
|
||||
private void EnqueueDeferred(AlarmConditionState condition, DeferredConditionReports queue, IFilterTarget e)
|
||||
{
|
||||
try
|
||||
queue.Pending.Enqueue(e);
|
||||
CountDeferred(+1, condition);
|
||||
while (queue.Pending.Count > 1
|
||||
&& (queue.Pending.Count > MaxDeferredEventsPerCondition || _deferredEventReports > MaxDeferredEventsTotal))
|
||||
{
|
||||
await report.ConfigureAwait(false);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogError(ex, "OtOpcUaNodeManager: failed to report Part 9 condition event for {NodeId}", source.NodeId);
|
||||
}
|
||||
finally
|
||||
{
|
||||
Interlocked.Decrement(ref _deferredEventReports);
|
||||
queue.Pending.Dequeue();
|
||||
CountDeferred(-1, condition);
|
||||
CountDiscarded("coalesced");
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>Delivers a condition's deferred events in order, one at a time, then forgets the queue.</summary>
|
||||
private async Task DrainDeferredAsync(AlarmConditionState condition, DeferredConditionReports queue, Task inFlight)
|
||||
{
|
||||
var current = inFlight;
|
||||
while (true)
|
||||
{
|
||||
try
|
||||
{
|
||||
await current.ConfigureAwait(false);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogError(ex, "OtOpcUaNodeManager: failed to report Part 9 condition event for {NodeId}", condition.NodeId);
|
||||
}
|
||||
|
||||
IFilterTarget next;
|
||||
lock (_eventReportGate)
|
||||
{
|
||||
CountDeferred(-1, condition);
|
||||
if (queue.Pending.Count == 0)
|
||||
{
|
||||
_deferredReports.Remove(condition);
|
||||
return;
|
||||
}
|
||||
|
||||
next = queue.Pending.Dequeue();
|
||||
}
|
||||
|
||||
try
|
||||
{
|
||||
current = DeliverConditionEventAsync(condition, condition, next).AsTask();
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
current = Task.FromException(ex);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// The SDK's <c>NodeState.ReportEventAsync</c> walk — the node's sync <c>OnReportEvent</c>, its async
|
||||
/// <c>OnReportEventAsync</c> (a <c>MonitoredNode2</c>), then each inverse notifier recursively from a snapshot of
|
||||
/// the notifier list, then (<c>BaseInstanceState</c>) the Parent — with a liveness check before every hop (#838
|
||||
/// review). A guard test pins that the SDK still has no other override. <c>EventReported</c> is not
|
||||
/// raised: nothing in the SDK or this repo subscribes to it, and it cannot be raised from outside the node.
|
||||
/// </summary>
|
||||
private async ValueTask DeliverConditionEventAsync(AlarmConditionState condition, NodeState node, IFilterTarget e)
|
||||
{
|
||||
if (!IsLiveCondition(condition))
|
||||
{
|
||||
// Removed or replaced while this event waited. Counted once, at the event's first hop.
|
||||
if (ReferenceEquals(node, condition)) CountDiscarded("not_live");
|
||||
return;
|
||||
}
|
||||
|
||||
node.OnReportEvent?.Invoke(SystemContext, node, e);
|
||||
if (node.OnReportEventAsync is { } handler)
|
||||
await handler(SystemContext, node, e, CancellationToken.None).ConfigureAwait(false);
|
||||
|
||||
var notifiers = new List<NodeState.Notifier>();
|
||||
node.GetNotifiers(SystemContext, notifiers);
|
||||
foreach (var notifier in notifiers)
|
||||
{
|
||||
if (notifier.IsInverse && notifier.Node is { } next)
|
||||
await DeliverConditionEventAsync(condition, next, e).ConfigureAwait(false);
|
||||
}
|
||||
|
||||
// BaseInstanceState.ReportEventAsync (the only override in SDK 2.0.1 — decompiled Types/Server/Core) runs the
|
||||
// base walk above and THEN reports to its Parent: that is how a condition's events reach the folder it hangs under.
|
||||
if (node is BaseInstanceState { Parent: { } parent })
|
||||
await DeliverConditionEventAsync(condition, parent, e).ConfigureAwait(false);
|
||||
}
|
||||
|
||||
/// <summary>Whether <paramref name="condition"/> is the instance the address space currently holds for its NodeId.</summary>
|
||||
private bool IsLiveCondition(NodeState condition) =>
|
||||
_alarmConditions.TryGetValue(MapKey(condition.NodeId), out var live) && ReferenceEquals(live, condition);
|
||||
|
||||
/// <summary>Adjusts the deferred depth and logs episode entry/exit. Under the gate.</summary>
|
||||
private void CountDeferred(int delta, AlarmConditionState condition)
|
||||
{
|
||||
var before = _deferredEventReports;
|
||||
_deferredEventReports += delta;
|
||||
OtOpcUaTelemetry.ConditionEventReportsDeferred.Add(delta);
|
||||
if (delta > 0) _episodeDeferred += delta;
|
||||
|
||||
if (before == 0 && delta > 0)
|
||||
{
|
||||
_episodeDeferred = delta;
|
||||
_episodeDiscarded = 0;
|
||||
var now = DateTime.UtcNow;
|
||||
_episodeWarned = now - _lastDeferralWarningUtc >= DeferralWarningInterval;
|
||||
if (_episodeWarned)
|
||||
{
|
||||
_lastDeferralWarningUtc = now;
|
||||
_logger.LogWarning(
|
||||
"OtOpcUaNodeManager: condition events are being deferred — an event notifier's delivery channel is "
|
||||
+ "full (first: {NodeId}). Folder event subscribers are not keeping up, or Lock is held across a large "
|
||||
+ "burst; events stay ordered per condition, and past {PerCondition} waiting per condition the oldest "
|
||||
+ "give way to the newest (Gitea #838)",
|
||||
condition.NodeId, MaxDeferredEventsPerCondition);
|
||||
}
|
||||
}
|
||||
else if (_deferredEventReports == 0 && before > 0 && _episodeWarned)
|
||||
{
|
||||
_logger.LogInformation(
|
||||
"OtOpcUaNodeManager: deferred condition events drained ({Deferred} deferred, {Discarded} discarded this episode)",
|
||||
_episodeDeferred, _episodeDiscarded);
|
||||
}
|
||||
}
|
||||
|
||||
private void CountDiscarded(string reason)
|
||||
{
|
||||
Interlocked.Increment(ref _discardedEventReports);
|
||||
lock (_eventReportGate) _episodeDiscarded++;
|
||||
OtOpcUaTelemetry.ConditionEventReportsDiscarded.Add(1, new KeyValuePair<string, object?>("reason", reason));
|
||||
}
|
||||
|
||||
/// <summary>One condition's deferred events: the one in flight is held by the drain loop, the rest wait here.</summary>
|
||||
private sealed class DeferredConditionReports
|
||||
{
|
||||
public readonly Queue<IFilterTarget> Pending = new();
|
||||
}
|
||||
|
||||
|
||||
/// <summary>
|
||||
/// The gate-relevant slice of a Part 9 condition's state — exactly the fields that drive a
|
||||
/// condition event AND that an <see cref="AlarmConditionSnapshot"/> can change. As a record, two
|
||||
@@ -5316,6 +5479,13 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3
|
||||
{
|
||||
if (_allEventsItems.IsEmpty) return;
|
||||
|
||||
// #838 review: never for a condition a deploy has removed or replaced. The hook stays attached to a torn-down
|
||||
// instance, and an event reported off Lock (a deferred report, an SDK method audit racing a teardown) would
|
||||
// otherwise hand Server-object subscribers a ghost Retain alarm after its NodeDeleted — and, with the ACL on,
|
||||
// a non-live scripted condition skips ConditionVisibilityScopes' "condition alone" rule and could be admitted
|
||||
// through its OLD equipment folder (#808).
|
||||
if (!IsLiveCondition(node)) return;
|
||||
|
||||
var scopes = AclEnforced ? ConditionVisibilityScopes(node) : null;
|
||||
var reportingSession = (context as ISessionSystemContext)?.SessionId;
|
||||
|
||||
|
||||
@@ -192,6 +192,28 @@ public sealed class ApiKeyAuthenticationTests
|
||||
auth.DescribeCredential("apikey", "not-a-key").ShouldBe("user:apikey");
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// #844 review — ConfigDbApiKeyStore matches the key id with SQL Server <c>=</c> (case-insensitive, trailing
|
||||
/// spaces ignored), so case / trailing-space variants of one key id reach the same key and must be ONE lockout
|
||||
/// identity; ids outside the minted lowercase-hex charset (no real key carries one) share a single identity.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public void Credential_identities_are_canonical_so_variants_of_one_key_id_share_a_count()
|
||||
{
|
||||
var auth = Auth(SetUp().Store);
|
||||
const string id = "0123456789abcdef";
|
||||
|
||||
auth.DescribeCredential("apikey", $"otua_{id}_s1").ShouldBe($"key:{id}");
|
||||
auth.DescribeCredential("apikey", $"otua_{id.ToUpperInvariant()}_s2").ShouldBe($"key:{id}");
|
||||
auth.DescribeCredential("apikey", $"otua_{id} _s3").ShouldBe($"key:{id}");
|
||||
auth.DescribeCredential("apikey", $"otua_ {id.ToUpperInvariant()} _s4").ShouldBe($"key:{id}");
|
||||
|
||||
auth.DescribeCredential("apikey", "otua_not-hex_s").ShouldBe("key:<not-a-minted-id>");
|
||||
auth.DescribeCredential("apikey", "otua_zzzz_s").ShouldBe(auth.DescribeCredential("apikey", "otua_yyyy_s"));
|
||||
|
||||
auth.DescribeCredential("APIKEY ", "garbage").ShouldBe("user:apikey");
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task An_unset_pepper_denies_rather_than_authenticating_everything()
|
||||
{
|
||||
|
||||
+61
@@ -104,6 +104,8 @@ public sealed class FolderEventFloodWireTests
|
||||
flooded.Count.ShouldBeLessThanOrEqualTo((int)onA.Status.QueueSize, "delivery is bounded by the item's queue, not the flood");
|
||||
flooded.ShouldBe(flooded.Order().ToList(), "events leave in the order they were reported");
|
||||
flooded[^1].ShouldBe(FloodSize - 1);
|
||||
rig.Nm.DiscardedEventReports.ShouldBeGreaterThan(0,
|
||||
"past the per-condition budget the oldest waiting events give way to the newest (coalesced)");
|
||||
|
||||
// And the node manager is live: a later event is decided by the per-folder ACL and delivered.
|
||||
rig.Project(native: true, State("after"));
|
||||
@@ -121,6 +123,65 @@ public sealed class FolderEventFloodWireTests
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// #838 review — events deferred behind a full channel must not outlive their condition. A deploy that removes
|
||||
/// the condition while its events wait used to let them reach Server-object (<c>i=2253</c>) subscribers after the
|
||||
/// NodeDeleted — a ghost Retain alarm for ScadaBridge — and the folder subscribers too. Every hop of a deferred
|
||||
/// report now re-checks that the condition is still the live instance.
|
||||
/// </summary>
|
||||
/// <param name="enforced">Per-folder / Server-object ACL on or off (off is the production default).</param>
|
||||
[Theory]
|
||||
[InlineData(true)]
|
||||
[InlineData(false)]
|
||||
public async Task Deferred_events_of_a_condition_removed_while_they_wait_are_never_delivered(bool enforced)
|
||||
{
|
||||
var ct = TestContext.Current.CancellationToken;
|
||||
await using var rig = await ServerEventRig.StartAsync("Folder838Ghost", encrypted: false, ct);
|
||||
rig.KeyRoles = [];
|
||||
rig.MaterialiseNative();
|
||||
rig.Enforce(enforced, [ServerEventRig.Key()], KeyEquipmentGrant(EqA));
|
||||
|
||||
using var keyed = await rig.ConnectAsync(anonymous: false, ct);
|
||||
var events = new RecordedEvents();
|
||||
var subscription = await RecordedEvents.SubscribeAsync(keyed, events, ct);
|
||||
var onA = RecordedEvents.AddEventItem(subscription, ServerEventRig.UnsNodeId(keyed, EqA));
|
||||
var onServer = RecordedEvents.AddEventItem(subscription, ObjectIds.Server);
|
||||
await subscription.ApplyChangesAsync(ct);
|
||||
onA.Status.Error.ShouldBeNull();
|
||||
onServer.Status.Error.ShouldBeNull();
|
||||
await subscription.SetPublishingModeAsync(false, ct);
|
||||
|
||||
// One Lock hold: flood past the channel (the tail is deferred), then the deploy removes the condition.
|
||||
await Task.Factory.StartNew(() =>
|
||||
{
|
||||
lock (rig.Nm.Lock)
|
||||
{
|
||||
for (var i = 0; i < FloodSize; i++) rig.Project(native: true, State($"f{i}"));
|
||||
rig.Nm.DeferredEventReports.ShouldBeGreaterThan(0, "the flood must actually have deferred events");
|
||||
rig.Nm.RemoveAlarmConditionNode(ServerEventRig.RawAlarmPath, AddressSpaceRealm.Raw).ShouldBeTrue();
|
||||
}
|
||||
}, ct, TaskCreationOptions.LongRunning, TaskScheduler.Default).WaitAsync(TimeSpan.FromSeconds(60), ct);
|
||||
|
||||
await RecordedEvents.WaitUntilAsync(() => rig.Nm.DeferredEventReports == 0);
|
||||
rig.Nm.DiscardedEventReports.ShouldBeGreaterThan(0, "the waiting events were dropped, not delivered");
|
||||
await Task.Delay(TimeSpan.FromSeconds(2), ct);
|
||||
|
||||
await subscription.SetPublishingModeAsync(true, ct);
|
||||
await Task.Delay(TimeSpan.FromSeconds(2), ct);
|
||||
int[] Flooded(MonitoredItem item) => events.Of(item)
|
||||
.Select(RecordedEvents.Message)
|
||||
.Where(m => m?.StartsWith('f') == true)
|
||||
.Select(m => int.Parse(m![1..], System.Globalization.CultureInfo.InvariantCulture))
|
||||
.ToArray();
|
||||
|
||||
// Positive control: the events reported before the channel filled did arrive (Server object: synchronously at
|
||||
// the condition's own hop).
|
||||
Flooded(onServer).ShouldNotBeEmpty();
|
||||
// None of the deferred tail — which was still waiting when the condition was removed — reached anyone.
|
||||
Flooded(onServer).Max().ShouldBeLessThan(FloodSize - 1, "no ghost reached the Server-object subscriber");
|
||||
Flooded(onA).DefaultIfEmpty(-1).Max().ShouldBeLessThan(FloodSize - 1, "no ghost reached the folder subscriber");
|
||||
}
|
||||
|
||||
private static AlarmConditionSnapshot State(string message) =>
|
||||
new(true, false, false, true, AlarmShelvingKind.Unshelved, 700, message);
|
||||
|
||||
|
||||
@@ -0,0 +1,41 @@
|
||||
using System.Reflection;
|
||||
using Opc.Ua;
|
||||
using Opc.Ua.Server;
|
||||
using Shouldly;
|
||||
using Xunit;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests;
|
||||
|
||||
/// <summary>
|
||||
/// Gitea #838 review: <c>OtOpcUaNodeManager.DeliverConditionEventAsync</c> re-implements the SDK's event walk
|
||||
/// (<c>NodeState.ReportEventAsync</c>, plus <c>BaseInstanceState</c>'s forward to its Parent) so it can re-check the
|
||||
/// condition's liveness before every hop. That copy is only faithful while no other node type changes the walk —
|
||||
/// missing the Parent hop alone silently stopped condition events reaching their folders. This pins the set of
|
||||
/// types that define <c>ReportEvent</c> / <c>ReportEventAsync</c> across the SDK and this server.
|
||||
/// </summary>
|
||||
public sealed class ConditionEventWalkGuardTests
|
||||
{
|
||||
[Fact]
|
||||
public void Only_NodeState_and_BaseInstanceState_define_the_event_walk()
|
||||
{
|
||||
Assembly[] assemblies =
|
||||
[
|
||||
typeof(NodeState).Assembly, // Opc.Ua.Types
|
||||
typeof(ApplicationConfiguration).Assembly, // Opc.Ua.Core
|
||||
typeof(StandardServer).Assembly, // Opc.Ua.Server
|
||||
typeof(OtOpcUaNodeManager).Assembly, // this server
|
||||
];
|
||||
|
||||
var definers = assemblies.Distinct()
|
||||
.SelectMany(a => a.GetTypes())
|
||||
.Where(t => typeof(NodeState).IsAssignableFrom(t))
|
||||
.Where(t => t.GetMethods(BindingFlags.Instance | BindingFlags.Public | BindingFlags.NonPublic | BindingFlags.DeclaredOnly)
|
||||
.Any(m => m.Name is nameof(NodeState.ReportEvent) or nameof(NodeState.ReportEventAsync)))
|
||||
.Select(t => t.FullName)
|
||||
.Order()
|
||||
.ToArray();
|
||||
|
||||
definers.ShouldBe(["Opc.Ua.BaseInstanceState", "Opc.Ua.NodeState"],
|
||||
"a new override changes the event walk — mirror it in OtOpcUaNodeManager.DeliverConditionEventAsync");
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user