Merge #854 ConditionRefresh not overtaken by deferred condition events (Server-object items)
v2-ci / unit-tests (push) Failing after 1h12m23s

This commit is contained in:
Joseph Doherty
2026-11-11 07:39:25 -05:00
5 changed files with 1278 additions and 55 deletions
+7 -2
View File
@@ -211,8 +211,13 @@ Two things to know before relying on any of this:
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 —
to newest); Warning/Information + `otopcua.opcua.condition_events.{deferred,discarded}`. A Server-object ConditionRefresh
(and ConditionRefresh2, #854) with nothing deferred is the plain walk OFF `Lock`; while deferring it takes a sequence
CUT (one short `Lock` hold: watermark + frozen snapshots of Retain conditions from `_alarmConditions`), queues the
replay off `Lock` one event per hold of a per-item gate, and withholds older deferred events from THAT item only
(re-sent condition, or Retain=true of one omitted because NOT retained — never one omitted by ACL). Item state is a
`ConditionalWeakTable` keyed by the item; cuts clear when the deferral count hits 0. Tests:
`ConditionRefreshDeferralWireTests`; folder items keep the residual (SDK channel + memoised verdict). 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`,
+48 -4
View File
@@ -882,15 +882,59 @@ Behaviours worth knowing before enabling it:
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.
- **`ConditionRefresh` is not overtaken by deferred events on Server-object items (Gitea #854).** A refresh of an
`i=2253` item replays each condition's current state between RefreshStart and RefreshEnd; the condition's deferred
events — all older — used to follow RefreshEnd, briefly showing a stale state or re-adding an alarm the refresh had
just dropped (ScadaBridge's snapshot swap drops absent conditions, then a stale active event re-added it). Every
condition event now carries its report sequence. Two paths, chosen per refresh:
- **Nothing deferred (the common case): unchanged and off `Lock`.** The SDK-shaped notifier walk, queued as live
condition nodes; `Lock` is taken only to copy the root-notifier list. It stands only if no deferral episode existed
at any moment of it (deferral count zero before and after, episode number unchanged); otherwise the cut path runs
too, and its replay — later in the same RefreshStart..RefreshEnd — supersedes anything that slipped in between.
- **Deferring: the cut.** ACL/role verdicts first, off `Lock`. Then ONE short `Lock` hold takes the sequence
watermark and a frozen snapshot of each Retain condition straight from the condition map (O(conditions), no tree
walk), and installs per item, under that item's gate, the **cut** plus a "replay pending" marker. The replay is
queued **off `Lock`**, one event per hold of the item gate. While the marker is set, every condition event to the
item takes the gate: one newer than the cut marks its condition so the older frozen entry is dropped instead of
following it; a deferred one at or below the cut is withheld **from that item only** when the refresh re-sent its
condition, or when it is `Retain=true` for a condition the refresh omitted **because it was not retained**. An
older `Retain=false` event of such a condition still arrives (it agrees with the omission), and a condition that was
retained but omitted for the item's **grants** is never withheld (a grant added before the drain lets its events
through; the ACL decides them at delivery as ever).
- The deferral queues, every other Server-object item and every folder item are untouched. A refresh waits for the
drain at most for one in-progress `QueueEvent` on that item (the gate), never for the drain itself; a reporter
holding `Lock` likewise waits at most for one. Each item keeps one gate/cut state for its lifetime (keyed by the
item, so an unsubscribe + resubscribe keeps it), and cuts are cleared when the deferral count returns to zero.
The custom refresh is refused (`ObjectDisposedException`) once the node manager is disposing, and disposal waits
for refreshes in flight, as the base's private operation guard does. Meter:
`otopcua.opcua.condition_events.superseded` (per item and event). ConditionRefresh and ConditionRefresh2, flag
on and off.
- **Measured** (3000 retained conditions, longest `Lock` wait another thread saw during one refresh): the first
#854 cut (`1622d07f`, everything under `Lock`) 95–106 ms idle / 102–127 ms deferring; now 0–3 ms idle /
23–32 ms deferring (the frozen snapshots).
- ⚠️ **Residual: folder (notifier) items.** A refresh of an item on an equipment/device folder can still be followed
by older events of a condition. Two sources, neither reachable per item: the SDK's own `MonitoredNode2` channel
(up to 4096 events already accepted before the refresh — this exists in the stock SDK too, and in normal operation
for the few events in transit), and our deferral queue, whose folder hop goes through that channel, where the SDK
memoises the per-item permission verdict by (item, event type, source) so there is no per-event, per-item hook. The
events stay in the condition's order and its newest arrives last, so the client converges. Discarding the queue
globally on refresh would drop transitions for every non-refreshing subscriber while leaving the channel backlog;
making refresh wait would block a service call on the stalled drain. ScadaBridge also ignores a transition older
than the one it holds for a condition.
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).
`ConditionRefreshDeferralWireTests` (#854) stalls a folder item's SDK lock so its channel fills without `Lock` held,
refreshes the Server-object item (ConditionRefresh and ConditionRefresh2, flag on/off, condition retained or cleared)
and pins that nothing older than the replay follows RefreshEnd while the non-refreshing folder item still receives the
whole tail (also across an unsubscribe + resubscribe of the item, and the cut is cleared after the drain); a
probe-driven test pins that a refresh waits for a delivery already past its check (fails without the item gate); an
ACL-on test pins that a condition omitted for want of a grant is not withheld once the grant lands; a 3000-condition
test measures the `Lock` hold and pins that neither path queues its replay under `Lock`; and a stress test runs
reports, refreshes and an intermittent stall concurrently (its stream never goes backwards outside a refresh, and
every refresh ends on the newest state delivered).
**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
@@ -172,6 +172,15 @@ public static class OtOpcUaTelemetry
Meter.CreateCounter<long>("otopcua.opcua.condition_events.discarded", unit: "{event}",
description: "Deferred condition events discarded before delivery (reason=coalesced|not_live).");
/// <summary>
/// Deferred condition events withheld from one Server-object (<c>i=2253</c>) item because a ConditionRefresh of that
/// item had already replayed a newer state (Gitea #854) — one per item and event. Every other subscriber still
/// receives them; this only stops a refresh being overtaken by older events.
/// </summary>
public static readonly Counter<long> ConditionEventDeliveriesSuperseded =
Meter.CreateCounter<long>("otopcua.opcua.condition_events.superseded", unit: "{delivery}",
description: "Deferred condition events withheld from a Server-object item whose ConditionRefresh already replayed a newer state.");
// ---------------- Convenience helpers ----------------
/// <summary>
@@ -1,5 +1,6 @@
using System.Collections.Concurrent;
using System.Globalization;
using System.Runtime.CompilerServices;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Logging.Abstractions;
using Opc.Ua;
@@ -620,6 +621,13 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3
{
if (disposing)
{
// #854 review: refuse new Server-object refreshes and let the ones in flight finish (the base's private
// NodeManagerOperation guard does this for its own paths).
_disposing = true;
Interlocked.MemoryBarrier(); // the flag write must not be reordered after the in-flight read (#854 re-review)
if (!SpinWait.SpinUntil(() => Volatile.Read(ref _refreshesInFlight) == 0, TimeSpan.FromSeconds(10)))
_logger.LogWarning("Node manager disposed with {Count} Server-object ConditionRefresh call(s) still in flight after 10 s",
Volatile.Read(ref _refreshesInFlight));
Volatile.Read(ref _dataPlaneAuthority).Changed -= OnAuthorityChanged;
_liveItems.Clear();
_eventItems.Clear();
@@ -1386,7 +1394,8 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3
// in MaterialiseAlarmCondition), whose fan-outs deliver it to direct notifier subscriptions; the
// condition's own OnReportEvent hands the same snapshot to the Server-object (all-events)
// subscriptions, per item, through the ACL (see OnConditionReportEvent).
var snapshot = new InstanceStateSnapshot();
var snapshot = new ConditionEventSnapshot(
Interlocked.Increment(ref _conditionEventSequence), alarm.NodeId, alarm.Retain?.Value == true);
snapshot.Initialize(SystemContext, alarm);
ReportEventWithoutBlocking(alarm, snapshot);
}
@@ -1434,6 +1443,14 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3
/// <summary>Deferred condition events discarded before delivery, any reason (test observation).</summary>
internal long DiscardedEventReports => Interlocked.Read(ref _discardedEventReports);
private long _refreshSupersededDeliveries;
/// <summary>
/// Deferred condition events withheld from a Server-object item because a ConditionRefresh on that item already
/// delivered a newer state (Gitea #854) — one per item and event (test observation).
/// </summary>
internal long RefreshSupersededDeliveries => Interlocked.Read(ref _refreshSupersededDeliveries);
/// <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).
@@ -1517,6 +1534,8 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3
/// <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)
{
// Delivered off Lock from here on, so a ConditionRefresh can overtake it (#854, see OnConditionReportEvent).
if (e is ConditionEventSnapshot snapshot) snapshot.Deferred = true;
queue.Pending.Enqueue(e);
CountDeferred(+1, condition);
while (queue.Pending.Count > 1
@@ -1543,19 +1562,30 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3
_logger.LogError(ex, "OtOpcUaNodeManager: failed to report Part 9 condition event for {NodeId}", condition.NodeId);
}
IFilterTarget next;
IFilterTarget? next = null;
var episodeEnded = false;
lock (_eventReportGate)
{
CountDeferred(-1, condition);
if (queue.Pending.Count == 0)
{
_deferredReports.Remove(condition);
return;
episodeEnded = _deferredEventReports == 0;
}
else
{
next = queue.Pending.Dequeue();
}
next = queue.Pending.Dequeue();
}
if (episodeEnded)
{
ClearStaleRefreshCuts();
return;
}
if (next is null) return;
try
{
current = DeliverConditionEventAsync(condition, condition, next).AsTask();
@@ -1630,7 +1660,14 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3
condition.NodeId, MaxDeferredEventsPerCondition);
}
}
else if (_deferredEventReports == 0 && before > 0 && _episodeWarned)
if (_deferredEventReports == 0 && before > 0)
{
// #854: every deferred event has been delivered, so no refresh cut taken during this episode can match
// anything any more; a new number makes them all stale at once (see RefreshCut.Episode).
_deferralEpisode++;
}
if (_deferredEventReports == 0 && before > 0 && _episodeWarned)
{
_logger.LogInformation(
"OtOpcUaNodeManager: deferred condition events drained ({Deferred} deferred, {Discarded} discarded this episode)",
@@ -1651,6 +1688,138 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3
public readonly Queue<IFilterTarget> Pending = new();
}
/// <summary>
/// The report order of condition events (#854): incremented under <c>Lock</c> by <see cref="ReportConditionEvent"/>,
/// read under <c>Lock</c> by a Server-object ConditionRefresh as its cut.
/// </summary>
private long _conditionEventSequence;
/// <summary>
/// A condition event as <see cref="ReportConditionEvent"/> reports it: the SDK's frozen snapshot, plus what a
/// Server-object ConditionRefresh needs to recognise it as superseded (Gitea #854). Still an
/// <see cref="InstanceStateSnapshot"/>, so the SDK's event identity, filtered-retain and queue handling treat it
/// exactly as before.
/// </summary>
/// <param name="sequence">Its place in report order (<see cref="_conditionEventSequence"/>).</param>
/// <param name="conditionId">The reporting condition's NodeId (not the instance: a cut must not pin a removed node).</param>
/// <param name="retain">The condition's <c>Retain</c> as reported.</param>
private sealed class ConditionEventSnapshot(long sequence, NodeId conditionId, bool retain) : InstanceStateSnapshot
{
public long Sequence { get; } = sequence;
public NodeId ConditionId { get; } = conditionId;
public bool Retain { get; } = retain;
/// <summary>
/// Set (under the event-report gate, before the drain can dequeue it) when the event waits in a deferral queue:
/// only such an event reaches Server-object items off <c>Lock</c>, which is the only way a refresh can be overtaken.
/// </summary>
public bool Deferred { get; set; }
}
/// <summary>
/// Per Server-object item state (#854): the gate that orders a deferred delivery to the item against a refresh of
/// the item, and the cut of the item's latest refresh while deferred events may still be older than it.
/// </summary>
private sealed class AllEventsItemState
{
/// <summary>Held across "is this deferred event superseded? → QueueEvent", and across a refresh's
/// "QueueEvent the replay → install the cut". Only ever taken innermost (nothing but the item's own lock under it).</summary>
public readonly object Gate = new();
/// <summary>Under <see cref="Gate"/>.</summary>
public RefreshCut? Cut;
/// <summary>
/// The refresh whose frozen replay is being queued to the item (0 = none). Set under <c>Lock</c> AND the gate
/// at the cut; cleared under the gate once queued. While set, every condition event delivered to the item
/// takes the gate, and one newer than the cut marks its condition in <see cref="NewerSinceCut"/> so the replay
/// does not follow it with an older state.
/// </summary>
public volatile int PendingReplay;
/// <summary>Under <see cref="Gate"/>: the watermark of <see cref="PendingReplay"/>.</summary>
public long PendingWatermark;
/// <summary>Under <see cref="Gate"/>: conditions an event newer than the pending replay already reached.</summary>
public readonly HashSet<NodeId> NewerSinceCut = [];
}
private int _refreshGeneration;
/// <summary>
/// One <see cref="AllEventsItemState"/> per Server-object item for the item's whole lifetime (#854 review): keyed by
/// the item object itself, so an unsubscribe + resubscribe of the same item between a refresh and the drain keeps its
/// cut and its gate, and a drain holding the state always holds the one a refresh installs into.
/// </summary>
private readonly ConditionalWeakTable<IEventMonitoredItem, AllEventsItemState> _allEventsItemStates = new();
/// <summary>Server-object items currently holding a refresh cut (test observation, #854 review).</summary>
internal int ActiveRefreshCuts
{
get
{
var n = 0;
foreach (var item in _allEventsItems.Keys)
{
if (!_allEventsItemStates.TryGetValue(item, out var state)) continue;
lock (state.Gate) n += state.Cut is null ? 0 : 1;
}
return n;
}
}
private AllEventsItemState AllEventsItemStateOf(IEventMonitoredItem item) =>
_allEventsItemStates.GetValue(item, static _ => new AllEventsItemState());
/// <summary>
/// Advanced (under the event-report gate) each time the deferral count returns to zero: the end of a deferral
/// episode, after which no event a cut could withhold is still undelivered (#854 review).
/// </summary>
private long _deferralEpisode;
/// <summary>What a Server-object item's latest ConditionRefresh conveyed (#854).</summary>
/// <param name="Watermark">Every condition event with a sequence at or below this was reported before the replay was read.</param>
/// <param name="Episode">The deferral episode the cut was taken in; once it ends the cut is stale and inert.</param>
/// <param name="Refreshed">The conditions the replay re-sent to THIS item.</param>
/// <param name="RetainedAtCut">Every condition that was Retain at the watermark, whether or not this item was sent it
/// (one set shared by every item of the refresh).</param>
private sealed record RefreshCut(long Watermark, long Episode, HashSet<NodeId> Refreshed, HashSet<NodeId> RetainedAtCut)
{
/// <summary>
/// Whether the refresh already told the item something at least as new as <paramref name="e"/>. A re-sent
/// condition: its replayed state supersedes every older event of it. A condition the replay omitted because it
/// was NOT retained: an older Retain=true event would wrongly re-add it, so it is withheld; an older Retain=false
/// event agrees with the omission and is delivered (a client that did not reconcile on RefreshEnd needs it). A
/// condition that WAS retained but was omitted for this item (ACL / role at refresh time): never withheld — the
/// refresh told the item nothing about it, so a grant added before the drain still lets its events through
/// (decided by the ACL at delivery, as ever).
/// </summary>
public bool Supersedes(ConditionEventSnapshot e) =>
e.Sequence <= Watermark
&& (Refreshed.Contains(e.ConditionId) || (e.Retain && !RetainedAtCut.Contains(e.ConditionId)));
}
/// <summary>
/// Drops every refresh cut taken in an episode that has ended (#854 review), so a cut does not outlive the events it
/// guards. Called by the drain after the deferral count reached zero, holding no lock; a cut installed concurrently
/// for the new episode is kept.
/// </summary>
private void ClearStaleRefreshCuts()
{
var current = Interlocked.Read(ref _deferralEpisode);
foreach (var item in _allEventsItems.Keys)
{
if (!_allEventsItemStates.TryGetValue(item, out var state)) continue;
lock (state.Gate)
{
if (state.Cut is { } cut && cut.Episode < current) state.Cut = null; // episodes only grow: never clear a newer cut (#854 re-review)
}
}
}
/// <summary>
/// The gate-relevant slice of a Part 9 condition's state — exactly the fields that drive a
@@ -5488,6 +5657,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3
var scopes = AclEnforced ? ConditionVisibilityScopes(node) : null;
var reportingSession = (context as ISessionSystemContext)?.SessionId;
var deferred = e is ConditionEventSnapshot { Deferred: true } d ? d : null;
foreach (var item in _allEventsItems.Keys)
{
@@ -5501,10 +5671,47 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3
continue;
}
item.QueueEvent(e);
// #854: a synchronous event (under Lock) needs the gate only while a refresh's replay is pending, and the
// pending marker is installed under Lock, so an item no refresh has touched has nothing to order. A deferred
// event always takes the gate — its state is created here if need be, so a refresh racing it shares it.
AllEventsItemState? state;
if (deferred is not null) state = AllEventsItemStateOf(item);
else if (!_allEventsItemStates.TryGetValue(item, out state) || state.PendingReplay == 0 || e is not ConditionEventSnapshot)
{
item.QueueEvent(e);
continue;
}
// A deferred event reaches this item off Lock, so a ConditionRefresh of the item may have replayed a newer
// state since it was reported; and while a refresh's frozen replay is being queued, an event newer than it must
// not be followed by it. Both are decided and queued under the item's gate, which the refresh also holds for
// each replayed event it queues — so a deferred event lands before the replay or is judged against the cut.
lock (state.Gate)
{
// A cut from an episode that has ended is stale (only waiting for ClearStaleRefreshCuts).
if (deferred is not null
&& state.Cut is { } cut && cut.Episode == Interlocked.Read(ref _deferralEpisode) && cut.Supersedes(deferred))
{
Interlocked.Increment(ref _refreshSupersededDeliveries);
OtOpcUaTelemetry.ConditionEventDeliveriesSuperseded.Add(1);
continue;
}
if (state.PendingReplay != 0 && e is ConditionEventSnapshot { } ce && ce.Sequence > state.PendingWatermark)
state.NewerSinceCut.Add(ce.ConditionId);
if (deferred is not null) DeferredServerObjectDeliveryProbe?.Invoke();
item.QueueEvent(e);
}
}
}
/// <summary>
/// Test probe (#854): runs on the drain thread between a deferred event's supersession check and its queueing to a
/// Server-object item, inside the item's gate — the window a refresh must not slip into. Null in production.
/// </summary>
internal Action? DeferredServerObjectDeliveryProbe { get; set; }
/// <summary>
/// The SDK's audit-event guard from <c>MonitoredNode2.OnReportEvent</c>, reproduced exactly: an audit
/// event reaches an item only while the server is auditing, and only over a SignAndEncrypt or
@@ -6068,57 +6275,311 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3
}
// A refresh on a Server-object item is the other way it learns about conditions — it is how a
// client connecting after a transition gets the current alarm picture — and the SDK base replays
// every condition under every root notifier to it after only its own role-permission check. With
// enforcement on, those items are refreshed here instead, through the same
// IsConditionVisibleTo decision live delivery uses; an event that names no
// condition is withheld. With it off, the base runs unchanged (its per-notifier repeats of one
// condition share an EventId, which the item's queue already collapses to one copy).
// Items on one of this manager's own notifiers keep the base behaviour: a stood-down item is off
// its notifier's fan-out, which the base already treats as "not refreshed".
// client connecting after a transition gets the current alarm picture. Those items are always refreshed
// here (RefreshServerObjectItems): with enforcement on through the same IsConditionVisibleTo decision live
// delivery uses, and in both modes so the refresh can cut off deferred events it supersedes (#854).
// Items on one of this manager's own notifiers keep the base behaviour with enforcement off (a stood-down
// item is off its notifier's fan-out, which the base already treats as "not refreshed"), and
// RefreshDirectItems with it on.
/// <inheritdoc />
public override ServiceResult ConditionRefresh(OperationContext context, IList<IEventMonitoredItem> monitoredItems)
{
if (!AclEnforced) return base.ConditionRefresh(context, monitoredItems);
var direct = new List<IEventMonitoredItem>();
var serverItems = new List<IEventMonitoredItem>();
foreach (var item in monitoredItems)
// The base's NodeManagerOperation guard is private; this is its equivalent for the custom paths (#854 review):
// refused once disposal starts, and Dispose waits for the ones in flight.
Interlocked.Increment(ref _refreshesInFlight);
try
{
if (item is null) continue;
(item.NodeId == ObjectIds.Server ? serverItems : direct).Add(item);
}
if (_disposing) throw new ObjectDisposedException(GetType().Name);
var systemContext = SystemContext.Copy(context);
if (direct.Count > 0) RefreshDirectItems(systemContext, direct);
if (serverItems.Count == 0) return ServiceResult.Good;
List<NodeState> notifiers;
lock (Lock) notifiers = RootNotifiers?.ToList() ?? [];
var events = new List<IFilterTarget>();
foreach (var notifier in notifiers) notifier.ConditionRefresh(systemContext, events, true);
// One snapshot per condition, with its scopes resolved once.
var conditions = new List<(IFilterTarget Event, List<NodeId> Scopes)>();
var seen = new HashSet<NodeState>(System.Collections.Generic.ReferenceEqualityComparer.Instance);
foreach (var e in events)
{
if (ConditionOf(e) is not { } condition || !seen.Add(condition)) continue;
conditions.Add((e, ConditionVisibilityScopes(condition)));
}
foreach (var item in serverItems)
{
foreach (var (e, scopes) in conditions)
var direct = new List<IEventMonitoredItem>();
var serverItems = new List<IEventMonitoredItem>();
foreach (var item in monitoredItems)
{
if (!IsConditionVisibleTo(item, scopes)) continue;
if (ServiceResult.IsBad(ValidateEventRolePermissions(item, e))) continue;
item.QueueEvent(e);
if (item is null) continue;
(item.NodeId == ObjectIds.Server ? serverItems : direct).Add(item);
}
var enforced = AclEnforced;
var systemContext = SystemContext.Copy(context);
if (direct.Count > 0)
{
if (enforced) RefreshDirectItems(systemContext, direct);
else base.ConditionRefresh(context, direct);
}
if (serverItems.Count > 0) RefreshServerObjectItems(systemContext, serverItems, enforced);
return ServiceResult.Good;
}
finally
{
Interlocked.Decrement(ref _refreshesInFlight);
}
}
private int _refreshesInFlight;
private volatile bool _disposing;
/// <summary>
/// Test probe (#854 review): called once per Server-object refresh with whether it took the cut path and whether
/// the calling thread held <c>Lock</c> while it queued the replay. Null in production.
/// </summary>
internal Action<bool, bool>? ServerObjectRefreshProbe { get; set; }
/// <summary>
/// ConditionRefresh (and ConditionRefresh2) for Server-object (<c>i=2253</c>) items: each condition under a root
/// notifier once, through the item's ACL decision when <paramref name="enforced"/> (#586) — and never overtaken by
/// a deferred event older than what it replays (Gitea #854).
/// </summary>
/// <param name="systemContext">The refresh's system context.</param>
/// <param name="items">The Server-object items being refreshed.</param>
/// <param name="enforced">Whether the data-plane ACL is enforced.</param>
/// <remarks>
/// <para>
/// <b>The overtaking (#854).</b> While a folder subscriber's channel is full, a condition's later events wait
/// in its deferral queue (#838) and reach Server-object items only when the drain gets to them, off
/// <c>Lock</c>. A refresh in that window replays the condition's CURRENT state between RefreshStart and
/// RefreshEnd; the waiting events, all older, then followed RefreshEnd — a stale state, or a condition the
/// refresh had just dropped re-added, until the newest drained.
/// </para>
/// <para>
/// <b>Common case — nothing deferred: unchanged, and off <c>Lock</c>.</b> The replay is the notifier walk the SDK
/// base does, queued as live condition nodes, with <c>Lock</c> taken only to copy the root-notifier list. It is
/// kept only if no deferral episode existed at any moment of it: the deferral count is zero before AND after and
/// the episode number (<see cref="_deferralEpisode"/>) did not move. Otherwise a deferred event may have reached
/// the item after the replay, so the cut path below runs as well — its second replay, queued later inside the
/// same RefreshStart/RefreshEnd, supersedes anything that slipped in between.
/// </para>
/// <para>
/// <b>Deferring — the cut.</b> In ONE hold of <c>Lock</c> (under which every condition event is reported): the
/// watermark (<see cref="_conditionEventSequence"/>), the episode, and a frozen snapshot of each Retain
/// condition read from <see cref="_alarmConditions"/> (O(conditions), no tree walk), then queued to each item
/// with its <see cref="RefreshCut"/> installed under the item's <see cref="AllEventsItemState.Gate"/>. The ACL /
/// role verdicts are computed BEFORE taking <c>Lock</c>, against the live conditions (they depend on the event
/// type and source, not state). The cut and a "replay pending" marker are installed per item under its
/// <see cref="AllEventsItemState.Gate"/> inside that <c>Lock</c> hold; the replay is then queued OFF
/// <c>Lock</c>, one event per hold of the gate. While the marker is set every condition event to the item takes
/// the gate: one newer than the cut marks its condition so its (older) frozen replay entry is dropped instead of
/// following it; a deferred one at or below the cut is judged against the cut (<see cref="RefreshCut.Supersedes"/>)
/// — it lands before the replay entry or not at all.
/// </para>
/// <para>
/// <b>Per item, never global.</b> Only the refreshed item stops receiving the superseded events; the deferral
/// queue, every other Server-object item and every folder item still get them, in order. A refresh waits for a
/// drain at most for one in-progress <c>QueueEvent</c> on that item (the gate), never for the drain itself.
/// Cuts die with their episode (<see cref="ClearStaleRefreshCuts"/>).
/// </para>
/// <para>
/// <b>Lock order</b> is <c>Lock</c> → item gate → the SDK item's own lock, the same as a synchronous report
/// (<c>Lock</c> → <see cref="OnConditionReportEvent"/>); the drain and the replay queueing take the item gate
/// without <c>Lock</c>, and nothing under an item gate takes <c>Lock</c> or the event-report gate. A gate is held
/// for one <c>QueueEvent</c> at a time, so a reporter holding <c>Lock</c> waits at most for one.
/// </para>
/// <para>
/// With enforcement off this replaces the SDK base's Server-object loop with the same result: the base queued
/// one copy per root notifier a condition is wired to, which the item's queue collapsed to one (they share an
/// EventId); here it is one per condition. A replayed event that names no condition is withheld only under
/// enforcement (fail closed), as before. The cut path replays from <see cref="_alarmConditions"/>, which holds
/// every condition this manager materialises (all of them sit under a root notifier); branches are replayed as
/// the SDK's <c>ConditionState.ConditionRefresh</c> lists them.
/// </para>
/// </remarks>
private void RefreshServerObjectItems(ServerSystemContext systemContext, List<IEventMonitoredItem> items, bool enforced)
{
long episodeBefore;
bool deferringBefore;
lock (_eventReportGate)
{
episodeBefore = _deferralEpisode;
deferringBefore = _deferredEventReports > 0;
}
if (!deferringBefore)
{
List<NodeState> notifiers;
lock (Lock) notifiers = RootNotifiers?.ToList() ?? [];
var events = new List<IFilterTarget>();
foreach (var notifier in notifiers) notifier.ConditionRefresh(systemContext, events, true);
var replay = DedupeReplay(events, enforced);
ServerObjectRefreshProbe?.Invoke(false, Monitor.IsEntered(Lock));
foreach (var item in items)
{
// A cut-path replay of this item still being queued by a concurrent refresh is older than this live one:
// stop it, so it cannot follow this replay with frozen states.
if (_allEventsItemStates.TryGetValue(item, out var pending))
{
lock (pending.Gate)
{
pending.PendingReplay = 0;
pending.NewerSinceCut.Clear();
}
}
foreach (var (e, _, scopes) in replay)
{
if (scopes is not null && !IsConditionVisibleTo(item, scopes)) continue;
if (ServiceResult.IsBad(ValidateEventRolePermissions(item, e))) continue;
item.QueueEvent(e);
}
}
lock (_eventReportGate)
{
if (_deferralEpisode == episodeBefore && _deferredEventReports == 0) return;
}
}
return ServiceResult.Good;
RefreshServerObjectItemsWithCut(systemContext, items, enforced);
}
/// <summary>One replay entry per condition (the first the walk produced), with its ACL scopes resolved once.</summary>
private List<(IFilterTarget Event, NodeState? Condition, List<NodeId>? Scopes)> DedupeReplay(
List<IFilterTarget> events, bool enforced)
{
var replay = new List<(IFilterTarget Event, NodeState? Condition, List<NodeId>? Scopes)>();
var seen = new HashSet<NodeState>(System.Collections.Generic.ReferenceEqualityComparer.Instance);
foreach (var e in events)
{
if (ConditionOf(e) is not { } condition)
{
if (!enforced) replay.Add((e, null, null));
continue;
}
if (!seen.Add(condition)) continue;
replay.Add((e, condition, enforced ? ConditionVisibilityScopes(condition) : null));
}
return replay;
}
/// <summary>The cut path of <see cref="RefreshServerObjectItems"/> — see there.</summary>
private void RefreshServerObjectItemsWithCut(ServerSystemContext systemContext, List<IEventMonitoredItem> items, bool enforced)
{
// ACL / role verdicts per (item, condition) before Lock: they depend on the event type, source and grants only.
List<AlarmConditionState> conditions;
lock (Lock) conditions = [.. _alarmConditions.Values];
var known = new HashSet<AlarmConditionState>(conditions, System.Collections.Generic.ReferenceEqualityComparer.Instance);
var allowed = new HashSet<(IEventMonitoredItem, AlarmConditionState)>();
foreach (var condition in conditions)
{
var scopes = enforced ? ConditionVisibilityScopes(condition) : null;
foreach (var item in items)
{
if (scopes is not null && !IsConditionVisibleTo(item, scopes)) continue;
if (ServiceResult.IsBad(ValidateEventRolePermissions(item, condition))) continue;
allowed.Add((item, condition));
}
}
var states = items.Select(AllEventsItemStateOf).ToList();
var perItem = new List<(IFilterTarget Event, AlarmConditionState Condition)>[items.Count];
var generation = Interlocked.Increment(ref _refreshGeneration);
// #854 re-review: a throw between installing the "replay pending" markers and finishing the replay must not leave
// a marker set (every sync event to that item would keep taking the gate until its next refresh).
try
{
lock (Lock)
{
var watermark = Interlocked.Read(ref _conditionEventSequence);
long episode;
bool deferring;
lock (_eventReportGate)
{
episode = _deferralEpisode;
deferring = _deferredEventReports > 0;
}
// Frozen at the watermark: the replay is exactly the state every event at or below it describes.
var retained = new HashSet<NodeId>();
var replay = new List<(IFilterTarget Event, AlarmConditionState Condition)>();
var branches = new List<IFilterTarget>();
foreach (var condition in _alarmConditions.Values)
{
if (condition.Retain?.Value != true) continue;
retained.Add(condition.NodeId);
branches.Clear();
condition.ConditionRefresh(SystemContext, branches, true);
foreach (var target in branches)
{
if (target is not BaseInstanceState instance) continue;
var snapshot = new InstanceStateSnapshot();
snapshot.Initialize(SystemContext, instance);
replay.Add((snapshot, condition));
}
}
for (var i = 0; i < items.Count; i++)
{
var item = items[i];
var entries = new List<(IFilterTarget Event, AlarmConditionState Condition)>();
var refreshed = new HashSet<NodeId>();
foreach (var (e, condition) in replay)
{
// A condition materialised since the verdicts were taken is decided here (rare).
if (!allowed.Contains((item, condition)))
{
if (known.Contains(condition)) continue;
if (enforced && !IsConditionVisibleTo(item, ConditionVisibilityScopes(condition))) continue;
if (ServiceResult.IsBad(ValidateEventRolePermissions(item, e))) continue;
}
entries.Add((e, condition));
refreshed.Add(condition.NodeId);
}
perItem[i] = entries;
var state = states[i];
lock (state.Gate)
{
state.Cut = deferring ? new RefreshCut(watermark, episode, refreshed, retained) : null;
state.PendingWatermark = watermark;
state.NewerSinceCut.Clear();
state.PendingReplay = generation;
}
}
}
// Queued off Lock, one replayed event per hold of the item gate: an event that reached the item since the cut is
// newer than the frozen replay of its condition, so that replay entry is dropped rather than queued after it (the
// newer event is already inside RefreshStart..RefreshEnd). A later refresh of the item supersedes this one.
ServerObjectRefreshProbe?.Invoke(true, Monitor.IsEntered(Lock));
for (var i = 0; i < items.Count; i++)
{
var item = items[i];
var state = states[i];
foreach (var (e, condition) in perItem[i])
{
lock (state.Gate)
{
if (state.PendingReplay != generation) break;
if (state.NewerSinceCut.Contains(condition.NodeId)) continue;
item.QueueEvent(e);
}
}
lock (state.Gate)
{
if (state.PendingReplay == generation)
{
state.PendingReplay = 0;
state.NewerSinceCut.Clear();
}
}
}
}
finally
{
foreach (var state in states)
{
lock (state.Gate)
{
if (state.PendingReplay != generation) continue;
state.PendingReplay = 0;
state.NewerSinceCut.Clear();
}
}
}
}
/// <summary>
@@ -0,0 +1,704 @@
using System.Collections;
using System.Diagnostics;
using System.Reflection;
using Opc.Ua;
using Opc.Ua.Client;
using Opc.Ua.Server;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
using ZB.MOM.WW.OtOpcUa.Configuration.Entities;
using ZB.MOM.WW.OtOpcUa.Configuration.Enums;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Core.Authorization;
using MonitoredItem = Opc.Ua.Client.MonitoredItem;
namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests;
/// <summary>
/// Gitea #854 (from the #838 review): while a folder subscriber's event channel is full, a condition's later events
/// are deferred in its per-condition queue. A <c>ConditionRefresh</c> on a Server-object (<c>i=2253</c>) item in that
/// window delivers the condition's CURRENT state between RefreshStart and RefreshEnd — and the deferred events, all
/// older than that state, used to reach the same item AFTER RefreshEnd, so a client briefly showed (or, for a
/// condition the refresh had dropped, re-added) a stale alarm. They are now withheld from the refreshed item;
/// every other subscriber still receives them, and the refresh never waits for the stalled drain.
/// </summary>
/// <remarks>
/// The overload is produced deterministically: the folder item's server-side <c>MonitoredItem</c> lock is held, so
/// the SDK's per-node delivery task parks in <c>QueueEvent</c>, its 4096-slot channel fills, and the condition's
/// later events are deferred — without holding the node manager's <c>Lock</c>, which the refresh needs.
/// </remarks>
public sealed class ConditionRefreshDeferralWireTests
{
private const string EqA = ServerEventRig.EqA;
// > 4096 (channel) + 1 (parked in the delivery task) + 1 (the in-flight deferred write), < 1024 pending: nothing coalesces.
private const int FloodSize = 5000;
/// <param name="enforced">Per-folder / Server-object ACL on or off (off is the production default).</param>
/// <param name="finalRetained">
/// Whether the condition's state at refresh time is still Retain (so the refresh re-sends it) or has cleared
/// (so the refresh omits it and its final Retain=false event must still arrive).
/// </param>
/// <param name="refresh2">ConditionRefresh2 on the single item instead of ConditionRefresh on the subscription.</param>
/// <param name="resubscribe">
/// The Server-object item is unsubscribed and resubscribed (the SDK drives this on a monitoring-mode change)
/// between the refresh and the drain: its cut must survive (#854 review — it lived in the subscription map entry).
/// </param>
[Theory]
[InlineData(false, true, false, false)]
[InlineData(true, true, false, false)]
[InlineData(false, false, false, false)]
[InlineData(true, false, false, false)]
[InlineData(false, true, true, false)]
[InlineData(true, false, true, false)]
[InlineData(false, true, false, true)]
[InlineData(true, false, false, true)]
public async Task Deferred_events_older_than_a_refresh_never_reach_the_refreshed_server_object_item_after_RefreshEnd(
bool enforced, bool finalRetained, bool refresh2, bool resubscribe)
{
var ct = TestContext.Current.CancellationToken;
await using var rig = await ServerEventRig.StartAsync("Refresh854", encrypted: false, ct);
rig.KeyRoles = [];
rig.MaterialiseNative();
rig.Enforce(enforced, [ServerEventRig.Key()], KeyEquipmentGrant(EqA));
// The folder subscriber whose channel the stall fills, on its own session so its parked item never touches the
// refreshing client's subscription.
using var folderSession = await rig.ConnectAsync(anonymous: false, ct);
var folderEvents = new RecordedEvents();
var folderSubscription = await RecordedEvents.SubscribeAsync(folderSession, folderEvents, ct);
var onA = RecordedEvents.AddEventItem(folderSubscription, ServerEventRig.UnsNodeId(folderSession, EqA));
await folderSubscription.ApplyChangesAsync(ct);
onA.Status.Error.ShouldBeNull();
using var serverSession = await rig.ConnectAsync(anonymous: false, ct);
var serverEvents = new RecordedEvents();
var serverSubscription = await RecordedEvents.SubscribeAsync(serverSession, serverEvents, ct);
var onServer = RecordedEvents.AddEventItem(serverSubscription, ObjectIds.Server);
await serverSubscription.ApplyChangesAsync(ct);
onServer.Status.Error.ShouldBeNull();
// Warm both paths (the folder item's verdict is cached on its first event).
rig.Project(native: true, Active("seed"));
await RecordedEvents.WaitUntilAsync(() => folderEvents.Has(onA, "seed") && serverEvents.Has(onServer, "seed"));
await folderSubscription.SetPublishingModeAsync(false, ct);
using (var stall = ItemLockStall.Hold(FindServerItem(rig.Nm, onA.Status.Id)))
{
for (var i = 0; i < FloodSize; i++) rig.Project(native: true, Active($"f{i}"));
rig.Nm.DeferredEventReports.ShouldBeGreaterThan(0, "the flood must actually have deferred events");
rig.Project(native: true, finalRetained ? Active("final") : Cleared("final"));
// The refresh is a service call: it must complete while the drain is still stalled.
var refresh = Stopwatch.StartNew();
if (refresh2)
{
await serverSession.CallAsync(
ObjectTypeIds.ConditionType, MethodIds.ConditionType_ConditionRefresh2, ct,
serverSubscription.Id, onServer.Status.Id);
}
else
{
await serverSession.CallAsync(
ObjectTypeIds.ConditionType, MethodIds.ConditionType_ConditionRefresh, ct, serverSubscription.Id);
}
refresh.Elapsed.ShouldBeLessThan(TimeSpan.FromSeconds(10), "ConditionRefresh must not wait for the stalled drain");
await RecordedEvents.WaitUntilAsync(() => serverEvents.HasType(onServer, ObjectTypeIds.RefreshEndEventType));
rig.Nm.DeferredEventReports.ShouldBeGreaterThan(0, "the deferred events were still waiting at RefreshEnd");
rig.Nm.ActiveRefreshCuts.ShouldBe(1);
if (resubscribe)
{
var serverItem = FindAllEventsItem(rig.Nm, onServer.Status.Id);
rig.Nm.SubscribeToAllEvents(null!, serverSubscription.Id, serverItem, unsubscribe: true);
rig.Nm.SubscribeToAllEvents(null!, serverSubscription.Id, serverItem, unsubscribe: false);
}
stall.Release();
}
await RecordedEvents.WaitUntilAsync(() => rig.Nm.DeferredEventReports == 0);
// #854 review: a cut does not outlive its deferral episode.
await RecordedEvents.WaitUntilAsync(() => rig.Nm.ActiveRefreshCuts == 0);
rig.Nm.ActiveRefreshCuts.ShouldBe(0, "the cut is cleared when the deferral count returns to zero");
// Positive control: a later transition is delivered (and everything deferred before it has been decided).
rig.Project(native: true, Active("after"));
await RecordedEvents.WaitUntilAsync(() => serverEvents.Has(onServer, "after"));
await Task.Delay(TimeSpan.FromMilliseconds(300), ct);
var stream = serverEvents.Of(onServer);
var start = stream.ToList().FindIndex(f => f[RecordedEvents.EventTypeField].GetNodeId() == ObjectTypeIds.RefreshStartEventType);
var end = stream.ToList().FindIndex(f => f[RecordedEvents.EventTypeField].GetNodeId() == ObjectTypeIds.RefreshEndEventType);
start.ShouldBeGreaterThanOrEqualTo(0);
end.ShouldBeGreaterThan(start);
var replayed = stream.Skip(start + 1).Take(end - start - 1).Select(RecordedEvents.Message).ToList();
var afterEnd = stream.Skip(end + 1)
.Where(f => f[RecordedEvents.AckedField].TryGetValue(out bool _))
.Select(RecordedEvents.Message)
.ToList();
if (finalRetained)
{
replayed.ShouldBe(["final"], "the refresh re-sends the condition's current state");
afterEnd.ShouldBe(["after"], "nothing older than the refreshed state may follow RefreshEnd");
}
else
{
replayed.ShouldBeEmpty("a cleared (Retain=false) condition is not part of a refresh");
// The stale Retain=true events would re-add an alarm the refresh just dropped; the final Retain=false one
// agrees with the refresh and must still arrive (a client that did not reconcile on RefreshEnd needs it).
afterEnd.ShouldBe(["final", "after"]);
stream.Last(f => RecordedEvents.Message(f) == "final")[RecordedEvents.RetainField].GetBoolean().ShouldBeFalse();
}
rig.Nm.RefreshSupersededDeliveries.ShouldBeGreaterThan(0);
// The folder subscriber did not refresh: it still receives the deferred events, in order, ending with the newest.
await folderSubscription.SetPublishingModeAsync(true, ct);
await RecordedEvents.WaitUntilAsync(() => folderEvents.Has(onA, "after"));
var folderFlood = folderEvents.Of(onA)
.Select(RecordedEvents.Message)
.Where(m => m?.StartsWith('f') == true && m != "final")
.Select(m => int.Parse(m![1..], System.Globalization.CultureInfo.InvariantCulture))
.ToList();
folderFlood.ShouldBe(folderFlood.Order().ToList());
folderFlood[^1].ShouldBe(FloodSize - 1, "the non-refreshing subscriber keeps the deferred tail");
folderEvents.Has(onA, "final").ShouldBeTrue();
}
/// <summary>
/// #854 review (ACL on): a condition the refresh omitted for the item because of its grants — not because it was
/// cleared — must not have its deferred events withheld: a grant added before the drain lets them through, and the
/// item must end on the condition's newest state (it would otherwise miss an active alarm entirely).
/// </summary>
[Fact]
public async Task Deferred_events_of_a_condition_the_refresh_omitted_for_want_of_a_grant_are_not_withheld()
{
var ct = TestContext.Current.CancellationToken;
await using var rig = await ServerEventRig.StartAsync("Refresh854Acl", encrypted: false, ct);
rig.KeyRoles = [];
rig.MaterialiseNative();
var anonymousOnA = ServerEventRig.EquipmentGrant(EqA, NodePermissions.ReadOnly);
rig.Enforce(true, [ServerEventRig.Key()], anonymousOnA);
// The stalled folder subscriber is anonymous (granted EqA); the refreshing key session holds no grant yet.
using var folderSession = await rig.ConnectAsync(anonymous: true, ct);
var folderEvents = new RecordedEvents();
var folderSubscription = await RecordedEvents.SubscribeAsync(folderSession, folderEvents, ct);
var onA = RecordedEvents.AddEventItem(folderSubscription, ServerEventRig.UnsNodeId(folderSession, EqA));
await folderSubscription.ApplyChangesAsync(ct);
using var serverSession = await rig.ConnectAsync(anonymous: false, ct);
var serverEvents = new RecordedEvents();
var serverSubscription = await RecordedEvents.SubscribeAsync(serverSession, serverEvents, ct);
var onServer = RecordedEvents.AddEventItem(serverSubscription, ObjectIds.Server);
await serverSubscription.ApplyChangesAsync(ct);
rig.Project(native: true, Active("seed"));
await RecordedEvents.WaitUntilAsync(() => folderEvents.Has(onA, "seed"));
serverEvents.Has(onServer, "seed").ShouldBeFalse("the key holds no grant yet");
await folderSubscription.SetPublishingModeAsync(false, ct);
using (var stall = ItemLockStall.Hold(FindServerItem(rig.Nm, onA.Status.Id)))
{
for (var i = 0; i < FloodSize; i++) rig.Project(native: true, Active($"f{i}"));
rig.Project(native: true, Active("final"));
rig.Nm.DeferredEventReports.ShouldBeGreaterThan(0);
await serverSession.CallAsync(
ObjectTypeIds.ConditionType, MethodIds.ConditionType_ConditionRefresh, ct, serverSubscription.Id);
await RecordedEvents.WaitUntilAsync(() => serverEvents.HasType(onServer, ObjectTypeIds.RefreshEndEventType));
// The grant arrives while the condition's events still wait.
rig.Enforce(true, [ServerEventRig.Key()], anonymousOnA, KeyEquipmentGrant(EqA));
stall.Release();
}
await RecordedEvents.WaitUntilAsync(() => rig.Nm.DeferredEventReports == 0);
await RecordedEvents.WaitUntilAsync(() => serverEvents.Has(onServer, "final"));
var stream = serverEvents.Of(onServer).ToList();
var end = stream.FindIndex(f => f[RecordedEvents.EventTypeField].GetNodeId() == ObjectTypeIds.RefreshEndEventType);
stream.Take(end).Count(f => f[RecordedEvents.AckedField].TryGetValue(out bool _))
.ShouldBe(0, "the refresh omitted the condition: no grant");
var after = stream.Skip(end + 1)
.Where(f => f[RecordedEvents.AckedField].TryGetValue(out bool _))
.Select(RecordedEvents.Message)
.ToList();
after.ShouldNotBeEmpty();
after[^1].ShouldBe("final", "the item ends on the newest state of a condition it was never refreshed");
after.Count(m => m!.StartsWith('f') && m != "final").ShouldBeGreaterThan(0, "its deferred events were delivered, not withheld");
}
/// <summary>
/// #854 review: the cut path queues its frozen replay OFF <c>Lock</c>. An event newer than the cut that reaches the
/// item in that window (here a synchronously delivered condition, reported from the probe between the cut and the
/// queueing) must not be followed by its condition's older frozen replay entry.
/// </summary>
[Fact]
public async Task An_event_newer_than_the_cut_is_never_followed_by_its_conditions_older_frozen_replay()
{
var ct = TestContext.Current.CancellationToken;
await using var rig = await ServerEventRig.StartAsync("Refresh854Pending", encrypted: false, ct);
rig.KeyRoles = [];
rig.MaterialiseNative();
rig.Enforce(false, [ServerEventRig.Key()], KeyEquipmentGrant(EqA));
// A second condition under the device folder only: no folder item, so its events are always synchronous.
const string Free = ServerEventRig.RawDeviceFolder + "/free";
rig.Sink.MaterialiseAlarmCondition(Free, ServerEventRig.RawDeviceFolder, "free", "OffNormalAlarm", 700, AddressSpaceRealm.Raw, isNative: true);
rig.Sink.WriteAlarmCondition(Free, Active("x1"), DateTime.UtcNow, AddressSpaceRealm.Raw);
using var folderSession = await rig.ConnectAsync(anonymous: false, ct);
var folderEvents = new RecordedEvents();
var folderSubscription = await RecordedEvents.SubscribeAsync(folderSession, folderEvents, ct);
var onA = RecordedEvents.AddEventItem(folderSubscription, ServerEventRig.UnsNodeId(folderSession, EqA));
await folderSubscription.ApplyChangesAsync(ct);
using var serverSession = await rig.ConnectAsync(anonymous: false, ct);
var serverEvents = new RecordedEvents();
var serverSubscription = await RecordedEvents.SubscribeAsync(serverSession, serverEvents, ct);
var onServer = RecordedEvents.AddEventItem(serverSubscription, ObjectIds.Server);
await serverSubscription.ApplyChangesAsync(ct);
rig.Project(native: true, Active("seed"));
await RecordedEvents.WaitUntilAsync(() => folderEvents.Has(onA, "seed") && serverEvents.Has(onServer, "seed"));
await folderSubscription.SetPublishingModeAsync(false, ct);
var injected = 0;
rig.Nm.ServerObjectRefreshProbe = (cut, _) =>
{
if (cut && Interlocked.Exchange(ref injected, 1) == 0)
rig.Sink.WriteAlarmCondition(Free, Active("x2"), DateTime.UtcNow, AddressSpaceRealm.Raw);
};
using (var stall = ItemLockStall.Hold(FindServerItem(rig.Nm, onA.Status.Id)))
{
for (var i = 0; i < FloodSize; i++) rig.Project(native: true, Active($"f{i}"));
rig.Nm.DeferredEventReports.ShouldBeGreaterThan(0, "the cut path runs only while something is deferred");
await serverSession.CallAsync(
ObjectTypeIds.ConditionType, MethodIds.ConditionType_ConditionRefresh, ct, serverSubscription.Id);
await RecordedEvents.WaitUntilAsync(() => serverEvents.HasType(onServer, ObjectTypeIds.RefreshEndEventType));
stall.Release();
}
rig.Nm.ServerObjectRefreshProbe = null;
Volatile.Read(ref injected).ShouldBe(1);
var stream = serverEvents.Of(onServer).ToList();
var end = stream.FindIndex(f => f[RecordedEvents.EventTypeField].GetNodeId() == ObjectTypeIds.RefreshEndEventType);
var free = stream.Take(end)
.Select(RecordedEvents.Message)
.Where(m => m is "x1" or "x2")
.ToList();
free.ShouldContain("x2");
free[^1].ShouldBe("x2", "the older frozen replay of the condition must not follow its newer event");
}
/// <summary>#854 review: the custom Server-object refresh is refused once the node manager is disposed, like the base's.</summary>
[Fact]
public async Task A_refresh_after_disposal_is_refused()
{
var ct = TestContext.Current.CancellationToken;
var rig = await ServerEventRig.StartAsync("Refresh854Dispose", encrypted: false, ct);
var nm = rig.Nm;
await rig.DisposeAsync();
nm.Dispose();
Should.Throw<ObjectDisposedException>(() => nm.ConditionRefresh(null!, Array.Empty<IEventMonitoredItem>()));
}
/// <summary>
/// The race the item gate closes, made deterministic: a deferred event is between its supersession check and its
/// queueing to the Server-object item when the refresh arrives. The refresh must wait for that one delivery (so
/// the event lands before the replay), not slip in between and let it follow RefreshEnd.
/// </summary>
[Fact]
public async Task A_refresh_waits_for_a_deferred_delivery_already_past_its_check_so_it_lands_before_the_replay()
{
var ct = TestContext.Current.CancellationToken;
await using var rig = await ServerEventRig.StartAsync("Refresh854Gate", encrypted: false, ct);
rig.KeyRoles = [];
rig.MaterialiseNative();
rig.Enforce(false, [ServerEventRig.Key()], KeyEquipmentGrant(EqA));
using var folderSession = await rig.ConnectAsync(anonymous: false, ct);
var folderEvents = new RecordedEvents();
var folderSubscription = await RecordedEvents.SubscribeAsync(folderSession, folderEvents, ct);
var onA = RecordedEvents.AddEventItem(folderSubscription, ServerEventRig.UnsNodeId(folderSession, EqA));
await folderSubscription.ApplyChangesAsync(ct);
using var serverSession = await rig.ConnectAsync(anonymous: false, ct);
var serverEvents = new RecordedEvents();
var serverSubscription = await RecordedEvents.SubscribeAsync(serverSession, serverEvents, ct);
var onServer = RecordedEvents.AddEventItem(serverSubscription, ObjectIds.Server);
await serverSubscription.ApplyChangesAsync(ct);
rig.Project(native: true, Active("seed"));
await RecordedEvents.WaitUntilAsync(() => folderEvents.Has(onA, "seed") && serverEvents.Has(onServer, "seed"));
await folderSubscription.SetPublishingModeAsync(false, ct);
using var entered = new ManualResetEventSlim();
using var proceed = new ManualResetEventSlim();
var probed = 0;
rig.Nm.DeferredServerObjectDeliveryProbe = () =>
{
if (Interlocked.Increment(ref probed) != 1) return;
entered.Set();
proceed.Wait(TimeSpan.FromSeconds(30));
};
using (ItemLockStall.Hold(FindServerItem(rig.Nm, onA.Status.Id)))
{
for (var i = 0; i < FloodSize; i++) rig.Project(native: true, Active($"f{i}"));
rig.Project(native: true, Active("final"));
}
// The drain resumes and parks on its first deferred delivery to the Server-object item, past the check.
entered.Wait(TimeSpan.FromSeconds(10), ct).ShouldBeTrue("the drain must reach the Server-object item");
var refresh = serverSession.CallAsync(
ObjectTypeIds.ConditionType, MethodIds.ConditionType_ConditionRefresh, ct, serverSubscription.Id);
await Task.Delay(TimeSpan.FromSeconds(1), ct);
// (The SDK may answer the Call before the refresh body runs, so the replay is observed on the event stream.)
serverEvents.HasType(onServer, ObjectTypeIds.RefreshEndEventType)
.ShouldBeFalse("the refresh must wait for the delivery already inside the item's gate");
proceed.Set();
await refresh.WaitAsync(TimeSpan.FromSeconds(10), ct);
await RecordedEvents.WaitUntilAsync(() => serverEvents.HasType(onServer, ObjectTypeIds.RefreshEndEventType));
rig.Nm.DeferredServerObjectDeliveryProbe = null;
await RecordedEvents.WaitUntilAsync(() => rig.Nm.DeferredEventReports == 0);
rig.Project(native: true, Active("after"));
await RecordedEvents.WaitUntilAsync(() => serverEvents.Has(onServer, "after"));
await Task.Delay(TimeSpan.FromMilliseconds(300), ct);
var stream = serverEvents.Of(onServer).ToList();
var start = stream.FindIndex(f => f[RecordedEvents.EventTypeField].GetNodeId() == ObjectTypeIds.RefreshStartEventType);
var end = stream.FindIndex(f => f[RecordedEvents.EventTypeField].GetNodeId() == ObjectTypeIds.RefreshEndEventType);
end.ShouldBeGreaterThan(start);
var window = stream.Skip(start + 1).Take(end - start - 1).Select(RecordedEvents.Message).ToList();
window.Count.ShouldBe(2, "the parked delivery, then the replay");
window[0]!.ShouldStartWith("f");
window[1].ShouldBe("final");
stream.Skip(end + 1)
.Where(f => f[RecordedEvents.AckedField].TryGetValue(out bool _))
.Select(RecordedEvents.Message)
.ShouldBe(["after"]);
}
/// <summary>
/// Race / stress: reports, repeated refreshes and an intermittently stalled drain all run at once. For the
/// refreshing Server-object item, no condition event delivered after a RefreshEnd may be older than the state that
/// refresh replayed, and the stream still ends on the newest state. Nothing deadlocks.
/// </summary>
/// <param name="enforced">Per-folder / Server-object ACL on or off.</param>
[Theory]
[InlineData(false)]
[InlineData(true)]
public async Task Concurrent_reports_refreshes_and_drains_never_deliver_a_pre_refresh_state_after_RefreshEnd(bool enforced)
{
var ct = TestContext.Current.CancellationToken;
await using var rig = await ServerEventRig.StartAsync("Refresh854Stress", encrypted: false, ct);
rig.KeyRoles = [];
rig.MaterialiseNative();
rig.Enforce(enforced, [ServerEventRig.Key()], KeyEquipmentGrant(EqA));
using var folderSession = await rig.ConnectAsync(anonymous: false, ct);
var folderEvents = new RecordedEvents();
var folderSubscription = await RecordedEvents.SubscribeAsync(folderSession, folderEvents, ct);
var onA = RecordedEvents.AddEventItem(folderSubscription, ServerEventRig.UnsNodeId(folderSession, EqA));
await folderSubscription.ApplyChangesAsync(ct);
using var serverSession = await rig.ConnectAsync(anonymous: false, ct);
var serverEvents = new RecordedEvents();
var serverSubscription = await RecordedEvents.SubscribeAsync(serverSession, serverEvents, ct);
var onServer = RecordedEvents.AddEventItem(serverSubscription, ObjectIds.Server);
onServer.QueueSize = 100_000; // keep every event so the ordering check sees the whole stream
await serverSubscription.ApplyChangesAsync(ct);
rig.Project(native: true, Active("s0"));
await RecordedEvents.WaitUntilAsync(() => folderEvents.Has(onA, "s0") && serverEvents.Has(onServer, "s0"));
await folderSubscription.SetPublishingModeAsync(false, ct);
var item = FindServerItem(rig.Nm, onA.Status.Id);
using var stop = CancellationTokenSource.CreateLinkedTokenSource(ct);
var counter = 0;
var reporter = Task.Factory.StartNew(() =>
{
while (!stop.IsCancellationRequested)
{
// Bursts under one Lock hold, like an apply pass, interleaved with single reports.
lock (rig.Nm.Lock)
{
for (var i = 0; i < 500; i++) rig.Project(native: true, Active($"s{Interlocked.Increment(ref counter)}"));
}
rig.Project(native: true, Active($"s{Interlocked.Increment(ref counter)}"));
Thread.Sleep(2);
}
}, stop.Token, TaskCreationOptions.LongRunning, TaskScheduler.Default);
var deferredSeen = 0;
var staller = Task.Factory.StartNew(() =>
{
while (!stop.IsCancellationRequested)
{
using (ItemLockStall.Hold(item))
{
Thread.Sleep(400);
if (rig.Nm.DeferredEventReports > 0) Interlocked.Increment(ref deferredSeen);
}
Thread.Sleep(100);
}
}, stop.Token, TaskCreationOptions.LongRunning, TaskScheduler.Default);
var refusedAsInProgress = 0;
for (var r = 0; r < 40; r++)
{
var call = r % 3 == 2
? serverSession.CallAsync(ObjectTypeIds.ConditionType, MethodIds.ConditionType_ConditionRefresh2, ct,
serverSubscription.Id, onServer.Status.Id)
: serverSession.CallAsync(ObjectTypeIds.ConditionType, MethodIds.ConditionType_ConditionRefresh, ct,
serverSubscription.Id);
try
{
await call.WaitAsync(TimeSpan.FromSeconds(20), ct);
}
catch (ServiceResultException ex) when (ex.StatusCode == StatusCodes.BadRefreshInProgress)
{
// The SDK answers the Call before the refresh body runs, so under load the previous refresh of this
// subscription may still be in progress; the SDK refuses the overlap (its rule, not ours).
refusedAsInProgress++;
}
await Task.Delay(TimeSpan.FromMilliseconds(80), ct);
}
await stop.CancelAsync();
await reporter.WaitAsync(TimeSpan.FromSeconds(30), CancellationToken.None);
await staller.WaitAsync(TimeSpan.FromSeconds(30), CancellationToken.None);
Volatile.Read(ref deferredSeen).ShouldBeGreaterThan(1, "the stress must actually have run with deferred events");
await RecordedEvents.WaitUntilAsync(() => rig.Nm.DeferredEventReports == 0);
var newest = $"s{Volatile.Read(ref counter)}";
await folderSubscription.SetPublishingModeAsync(true, ct);
await RecordedEvents.WaitUntilAsync(() => serverEvents.Has(onServer, newest));
await Task.Delay(TimeSpan.FromMilliseconds(500), ct);
// Walk the Server-object stream (one condition). Outside a refresh its states never go backwards; and each refresh
// ends on the newest state the item has seen — its replay is never older than an event already delivered, and
// nothing older than it follows RefreshEnd.
var max = -1;
var last = -1;
var replaying = false;
var violations = new List<string>();
foreach (var f in serverEvents.Of(onServer))
{
var type = f[RecordedEvents.EventTypeField].GetNodeId();
if (type == ObjectTypeIds.RefreshStartEventType)
{
replaying = true;
continue;
}
if (type == ObjectTypeIds.RefreshEndEventType)
{
replaying = false;
if (last < max) violations.Add($"a refresh ended on s{last} after s{max} had been delivered");
continue;
}
if (!f[RecordedEvents.AckedField].TryGetValue(out bool _)) continue;
var n = int.Parse(RecordedEvents.Message(f)![1..], System.Globalization.CultureInfo.InvariantCulture);
if (!replaying && n < max) violations.Add($"s{n} delivered after s{max}");
max = Math.Max(max, n);
last = n;
}
violations.ShouldBeEmpty();
refusedAsInProgress.ShouldBeLessThan(40, "most refreshes must actually have run");
rig.Nm.RefreshSupersededDeliveries.ShouldBeGreaterThan(0, "some refresh must actually have overtaken deferred events");
RecordedEvents.Message(serverEvents.Of(onServer).Last(f => f[RecordedEvents.AckedField].TryGetValue(out bool _)))
.ShouldBe(newest, "the stream still converges on the newest state");
await RecordedEvents.WaitUntilAsync(() => folderEvents.Has(onA, newest));
}
/// <summary>
/// Review of 1622d07f: a Server-object refresh must not hold the node manager's <c>Lock</c> across a walk of a large
/// address space — ScadaBridge refreshes on every reconnect, and any session may loop ConditionRefresh. Measures
/// the longest any other thread waits for <c>Lock</c> while a refresh of 3000 retained conditions runs, with
/// nothing deferred (the common case) and with a deferral in progress, and pins that the common case does not
/// build or queue its replay under <c>Lock</c>.
/// </summary>
/// <param name="deferring">Whether a deferral episode is in progress when the refresh runs.</param>
[Theory]
[InlineData(false)]
[InlineData(true)]
public async Task A_refresh_of_a_large_address_space_keeps_the_node_manager_lock_hold_short(bool deferring)
{
const int Conditions = 3000;
var ct = TestContext.Current.CancellationToken;
await using var rig = await ServerEventRig.StartAsync("Refresh854Lock", encrypted: false, ct);
rig.KeyRoles = [];
rig.MaterialiseNative();
rig.Enforce(false, [ServerEventRig.Key()], KeyEquipmentGrant(EqA));
for (var i = 0; i < Conditions; i++)
{
var id = $"{ServerEventRig.RawDeviceFolder}/c{i}";
rig.Sink.MaterialiseAlarmCondition(id, ServerEventRig.RawDeviceFolder, $"c{i}", "OffNormalAlarm", 700, AddressSpaceRealm.Raw, isNative: true);
rig.Sink.WriteAlarmCondition(id, Active($"c{i}"), DateTime.UtcNow, AddressSpaceRealm.Raw);
}
using var folderSession = await rig.ConnectAsync(anonymous: false, ct);
var folderEvents = new RecordedEvents();
var folderSubscription = await RecordedEvents.SubscribeAsync(folderSession, folderEvents, ct);
var onA = RecordedEvents.AddEventItem(folderSubscription, ServerEventRig.UnsNodeId(folderSession, EqA));
await folderSubscription.ApplyChangesAsync(ct);
using var serverSession = await rig.ConnectAsync(anonymous: false, ct);
var serverEvents = new RecordedEvents();
var serverSubscription = await RecordedEvents.SubscribeAsync(serverSession, serverEvents, ct);
var onServer = RecordedEvents.AddEventItem(serverSubscription, ObjectIds.Server);
onServer.QueueSize = 10_000;
await serverSubscription.ApplyChangesAsync(ct);
rig.Project(native: true, Active("seed"));
await RecordedEvents.WaitUntilAsync(() => folderEvents.Has(onA, "seed") && serverEvents.Has(onServer, "seed"));
await folderSubscription.SetPublishingModeAsync(false, ct);
var stall = deferring ? ItemLockStall.Hold(FindServerItem(rig.Nm, onA.Status.Id)) : null;
try
{
if (deferring)
{
for (var i = 0; i < FloodSize; i++) rig.Project(native: true, Active($"f{i}"));
rig.Nm.DeferredEventReports.ShouldBeGreaterThan(0);
}
var probes = new System.Collections.Concurrent.ConcurrentQueue<(bool Cut, bool LockHeld)>();
rig.Nm.ServerObjectRefreshProbe = (cut, held) => probes.Enqueue((cut, held));
var waits = new List<double>();
for (var round = 0; round < 5; round++)
{
var ends = serverEvents.Of(onServer).Count(f => f[RecordedEvents.EventTypeField].GetNodeId() == ObjectTypeIds.RefreshEndEventType);
using var stop = new CancellationTokenSource();
var maxWait = 0.0;
var prober = Task.Factory.StartNew(() =>
{
var sw = new Stopwatch();
while (!stop.IsCancellationRequested)
{
sw.Restart();
lock (rig.Nm.Lock) { }
maxWait = Math.Max(maxWait, sw.Elapsed.TotalMilliseconds);
Thread.Yield();
}
}, CancellationToken.None, TaskCreationOptions.LongRunning, TaskScheduler.Default);
await serverSession.CallAsync(
ObjectTypeIds.ConditionType, MethodIds.ConditionType_ConditionRefresh, ct, serverSubscription.Id);
await RecordedEvents.WaitUntilAsync(() => serverEvents.Of(onServer)
.Count(f => f[RecordedEvents.EventTypeField].GetNodeId() == ObjectTypeIds.RefreshEndEventType) > ends);
await stop.CancelAsync();
await prober;
waits.Add(maxWait);
}
TestContext.Current.TestOutputHelper?.WriteLine(
$"#854 lock-hold deferring={deferring} conditions={Conditions}: max Lock wait per refresh (ms) = "
+ string.Join(", ", waits.Select(w => w.ToString("F1", System.Globalization.CultureInfo.InvariantCulture))));
// Neither path queues its replay under Lock; only a deferral makes a refresh take the cut path.
probes.ShouldNotBeEmpty();
probes.ShouldAllBe(p => !p.LockHeld);
if (deferring) probes.ShouldContain(p => p.Cut);
else probes.ShouldAllBe(p => !p.Cut);
}
finally
{
stall?.Dispose();
}
}
/// <summary>The server-side Server-object (all-events) item behind a client item.</summary>
private static IEventMonitoredItem FindAllEventsItem(OtOpcUaNodeManager nm, uint id)
{
var map = (IDictionary)typeof(OtOpcUaNodeManager)
.GetField("_allEventsItems", BindingFlags.Instance | BindingFlags.NonPublic)!
.GetValue(nm)!;
foreach (var key in map.Keys)
{
if (key is IEventMonitoredItem item && item.Id == id) return item;
}
throw new InvalidOperationException($"no server-side all-events item {id}");
}
/// <summary>The server-side event item behind a client item, from the node manager's direct-item map.</summary>
private static IEventMonitoredItem FindServerItem(OtOpcUaNodeManager nm, uint id)
{
var map = (IDictionary)typeof(OtOpcUaNodeManager)
.GetField("_eventItems", BindingFlags.Instance | BindingFlags.NonPublic)!
.GetValue(nm)!;
foreach (var key in map.Keys)
{
if (key is IEventMonitoredItem item && item.Id == id) return item;
}
throw new InvalidOperationException($"no server-side event item {id}");
}
private static AlarmConditionSnapshot Active(string message) =>
new(true, false, false, true, AlarmShelvingKind.Unshelved, 700, message);
private static AlarmConditionSnapshot Cleared(string message) =>
new(false, true, true, true, AlarmShelvingKind.Unshelved, 700, message);
private static NodeAcl KeyEquipmentGrant(string equipment) => new()
{
NodeAclId = $"acl-key-{equipment}",
ClusterId = ServerEventRig.Cluster,
Principal = ServerEventRig.KeyPrincipal,
ScopeKind = NodeAclScopeKind.Equipment,
ScopeId = $"eq-{equipment}",
PermissionFlags = NodePermissions.ReadOnly,
};
/// <summary>
/// Holds a server-side monitored item's private lock on a dedicated thread, so the SDK's per-node delivery task
/// parks in <c>QueueEvent</c> for that item (a stalled subscriber) without the node manager's <c>Lock</c> held.
/// </summary>
private sealed class ItemLockStall : IDisposable
{
private readonly ManualResetEventSlim _release = new();
private readonly Thread _thread;
private ItemLockStall(IEventMonitoredItem item)
{
var gate = (Lock)typeof(Opc.Ua.Server.MonitoredItem)
.GetField("m_lock", BindingFlags.Instance | BindingFlags.NonPublic)!
.GetValue(item)!;
using var held = new ManualResetEventSlim();
_thread = new Thread(() =>
{
gate.Enter();
try
{
held.Set();
_release.Wait();
}
finally { gate.Exit(); }
}) { IsBackground = true };
_thread.Start();
held.Wait();
}
public static ItemLockStall Hold(IEventMonitoredItem item) => new(item);
public void Release() => _release.Set();
public void Dispose()
{
_release.Set();
_thread.Join();
_release.Dispose();
}
}
}