dohertj2/lmxopcua
1
0
Fork 0
Code Issues 5 Pull Requests Actions Packages Projects Releases Wiki Activity

fix(opcua): Server-object ConditionRefresh off Lock unless deferring; cut path O(conditions) with replay queued off Lock (#854 review)

Browse Source 
Review of 1622d07f:
1. Perf: every i=2253 refresh ran the whole notifier walk + permission checks +
   QueueEvent under Lock. Now, with nothing deferred (the common case), the refresh is
   the SDK-shaped walk off Lock (Lock only to copy the root notifiers), kept only if no
   deferral episode existed at any moment of it (count 0 before and after, episode
   unchanged); otherwise the cut path also runs and its later replay supersedes. The
   cut path takes ACL/role verdicts off Lock, then one short Lock hold: watermark +
   frozen snapshots of Retain conditions from _alarmConditions (no tree walk) + per-item
   cut and "replay pending" marker; the replay is queued OFF Lock, one event per hold of
   the item gate. While pending, a condition event newer than the cut marks its
   condition so the older frozen entry is dropped rather than following it. A
   concurrent fast-path refresh cancels a pending cut replay of the same item.
   Measured, 3000 retained conditions, longest Lock wait seen by another thread during
   one refresh: 1622d07f 95-106 ms idle / 102-127 ms deferring; now 0-3 ms idle /
   23-32 ms deferring.
2. ACL: a retained condition omitted from the replay for want of a grant is never
   withheld (RefreshCut.RetainedAtCut); only Retain=true events of a condition omitted
   because it was NOT retained are.
3. One gate/cut state per item for its lifetime (ConditionalWeakTable keyed by the
   item): an unsubscribe + resubscribe between refresh and drain keeps the cut.
4. Cuts carry their deferral episode; ClearStaleRefreshCuts runs when the count
   returns to zero, and a stale-episode cut is inert.
5. Dispose guard for the custom refresh (ObjectDisposedException once disposing;
   Dispose waits for refreshes in flight).
6. Docs: a refresh waits at most for one in-progress QueueEvent on the item.

Tests: resubscribe variants, ACL-grant-lands test, pending-replay probe test,
dispose test, 3000-condition Lock-hold measurement with a probe pinning that neither
path queues its replay under Lock; stress invariant tightened (never backwards outside a
refresh; each refresh ends on the newest delivered state) and tolerant of the SDK's own
BadRefreshInProgress overlap refusal. Each new test fails against its targeted mutation.
This commit is contained in:
Joseph Doherty
2026-11-11 07:28:08 -05:00
parent 1622d07f2e
commit fe48458a41
4 changed files with 688 additions and 122 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.claude/rules/data-plane-auth.md
+6 -3
View File
@@ -212,9 +212,12 @@ Two things to know before relying on any of this:
`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}`. A Server-object ConditionRefresh
(and ConditionRefresh2) takes a sequence CUT in one `Lock` hold and withholds older deferred events from THAT item
only (re-sent condition, or a Retain=true event of an omitted one) under a per-item gate held across check+QueueEvent
(#854, `ConditionRefreshDeferralWireTests`); folder items keep the residual (SDK channel + memoised verdict). Folder events follow Part 3's shape —
(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`,
docs/security.md
+33 -12
View File
@@ -885,16 +885,33 @@ Behaviours worth knowing before enabling it:
- **`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). Now every
condition event carries its report sequence; the refresh takes its **cut** (the sequence, whether anything is
deferred, the replay, and its queueing) in **one hold of `Lock`**, so the replay is exactly the state at the cut, and
installs it per item under that item's gate, which a deferred delivery also holds across its check and
`QueueEvent`. A deferred event 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; an older `Retain=false` event of an
omitted condition still arrives (it agrees with the omission, and a client that does not reconcile on RefreshEnd
needs it). The deferral queues, every other Server-object item and every folder item are untouched; the refresh never
waits for a drain. Meter: `otopcua.opcua.condition_events.superseded` (per item and event). Both ConditionRefresh and
ConditionRefresh2, flag on and off.
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
@@ -912,8 +929,12 @@ Behaviours worth knowing before enabling it:
`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; a probe-driven test pins that a refresh waits for a delivery already past its check (fails without the
item gate); and a stress test runs reports, refreshes and an intermittent stall concurrently.
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
src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs
+350 -82
View File
@@ -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,10 @@ 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;
SpinWait.SpinUntil(() => Volatile.Read(ref _refreshesInFlight) == 0, TimeSpan.FromSeconds(10));
Volatile.Read(ref _dataPlaneAuthority).Changed -= OnAuthorityChanged;
_liveItems.Clear();
_eventItems.Clear();
@@ -1554,19 +1559,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();
@@ -1641,7 +1657,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)",
@@ -1704,21 +1727,94 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3
/// <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="Refreshed">The conditions the replay re-sent to the item; the others were Retain=false (or not visible).</param>
private sealed record RefreshCut(long Watermark, HashSet<NodeId> Refreshed)
/// <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's current state supersedes every older event of it; for a condition the replay OMITTED (it was not
/// retained), an older Retain=true event would wrongly re-add it, while an older Retain=false event agrees with
/// the omission and is still delivered (a client that did not reconcile on RefreshEnd needs it).
/// 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 && (e.Retain || Refreshed.Contains(e.ConditionId));
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;
}
}
}
@@ -5362,7 +5458,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3
/// unsubscribe arms, which the SDK's <c>MasterNodeManager</c> drives for every create, modify,
/// transfer, restore and delete of an all-events item, subscription teardown included.
/// </remarks>
private readonly ConcurrentDictionary<IEventMonitoredItem, AllEventsItemState> _allEventsItems = new();
private readonly ConcurrentDictionary<IEventMonitoredItem, byte> _allEventsItems = new();
// The SDK's base implementation attaches a Server-object item to the event
// fan-out of every one of this manager's root notifiers, so every alarm in the server reached it
@@ -5384,7 +5480,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3
lock (Lock)
{
if (unsubscribe) _allEventsItems.TryRemove(monitoredItem, out _);
else _allEventsItems.GetOrAdd(monitoredItem, static _ => new AllEventsItemState());
else _allEventsItems[monitoredItem] = 0;
ReconcileAllEventsMonitoring();
}
@@ -5560,7 +5656,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3
var reportingSession = (context as ISessionSystemContext)?.SessionId;
var deferred = e is ConditionEventSnapshot { Deferred: true } d ? d : null;
foreach (var (item, state) in _allEventsItems)
foreach (var item in _allEventsItems.Keys)
{
if (e is AuditEventState && !IsAuditDeliverableTo(item)) continue;
if (scopes is not null && !IsConditionVisibleTo(item, scopes)) continue;
@@ -5572,25 +5668,36 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3
continue;
}
if (deferred is null)
// #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;
}
// #854: a deferred event reaches this item off Lock, so a ConditionRefresh of the item may have replayed a
// newer state since it was reported. Decided and queued under the item's gate, which the refresh holds while
// it queues its replay and installs its cut — so it lands either before the replay or not at all.
// 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)
{
if (state.Cut?.Supersedes(deferred) == true)
// 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;
}
DeferredServerObjectDeliveryProbe?.Invoke();
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);
}
}
@@ -6174,26 +6281,47 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3
/// <inheritdoc />
public override ServiceResult ConditionRefresh(OperationContext context, IList<IEventMonitoredItem> 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 enforced = AclEnforced;
var systemContext = SystemContext.Copy(context);
if (direct.Count > 0)
var direct = new List<IEventMonitoredItem>();
var serverItems = new List<IEventMonitoredItem>();
foreach (var item in monitoredItems)
{
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
{
if (enforced) RefreshDirectItems(systemContext, direct);
else base.ConditionRefresh(context, direct);
Interlocked.Decrement(ref _refreshesInFlight);
}
if (serverItems.Count > 0) RefreshServerObjectItems(systemContext, serverItems, enforced);
return ServiceResult.Good;
}
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
@@ -6211,84 +6339,224 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3
/// refresh had just dropped re-added, until the newest drained.
/// </para>
/// <para>
/// <b>The cut.</b> Everything — the watermark (<see cref="_conditionEventSequence"/>), whether any event is
/// deferred, the replay itself and its queueing — happens in ONE hold of <c>Lock</c>, under which every condition
/// event is reported. So the replay is exactly the state at the watermark: an event at or below it is not newer
/// than the replay, one above it was reported after it. Each item's replay is queued and its
/// <see cref="RefreshCut"/> installed under the item's <see cref="AllEventsItemState.Gate"/>, which a deferred
/// delivery to the item also holds (<see cref="OnConditionReportEvent"/>): a deferred event therefore lands
/// before the replay, or is judged against the cut — withheld if superseded (<see cref="RefreshCut.Supersedes"/>).
/// Only events marked <see cref="ConditionEventSnapshot.Deferred"/> are judged: any other was delivered to
/// Server-object items synchronously under <c>Lock</c>, so it cannot follow a replay it predates.
/// <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. The refresh never
/// waits for a drain. No cut is kept while nothing is deferred (the common case).
/// 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 takes the item gate without <c>Lock</c>, and
/// nothing under an item gate takes <c>Lock</c> or the event-report gate.
/// (<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.
/// 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)
{
lock (Lock)
long episodeBefore;
bool deferringBefore;
lock (_eventReportGate)
{
var watermark = Interlocked.Read(ref _conditionEventSequence);
bool deferring;
lock (_eventReportGate) deferring = _deferredEventReports > 0;
episodeBefore = _deferralEpisode;
deferringBefore = _deferredEventReports > 0;
}
if (!deferringBefore)
{
List<NodeState> notifiers;
lock (Lock) notifiers = RootNotifiers?.ToList() ?? [];
var events = new List<IFilterTarget>();
foreach (var notifier in RootNotifiers?.ToList() ?? []) notifier.ConditionRefresh(systemContext, events, true);
// One event per condition, with its scopes resolved once.
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));
}
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)
{
var queued = new List<IFilterTarget>();
var refreshed = new HashSet<NodeId>();
foreach (var (e, condition, scopes) in replay)
// 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;
queued.Add(e);
if (condition is not null) refreshed.Add(condition.NodeId);
item.QueueEvent(e);
}
}
if (!_allEventsItems.TryGetValue(item, out var state))
lock (_eventReportGate)
{
if (_deferralEpisode == episodeBefore && _deferredEventReports == 0) return;
}
}
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);
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)
{
// Not (or no longer) subscribed: no deferred delivery can reach it, so there is nothing to cut.
foreach (var e in queued) item.QueueEvent(e);
continue;
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)
{
foreach (var e in queued) item.QueueEvent(e);
// A cut is only needed while an event at or below the watermark may still be delivered. Refreshes are
// serialised by Lock, so a later one always carries the later watermark and simply replaces it.
state.Cut = deferring ? new RefreshCut(watermark, refreshed) : null;
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();
}
}
}
tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/ConditionRefreshDeferralWireTests.cs
+299 -25
View File
@@ -9,6 +9,7 @@ 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;
@@ -40,15 +41,21 @@ public sealed class ConditionRefreshDeferralWireTests
/// (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)]
[InlineData(true, true, false)]
[InlineData(false, false, false)]
[InlineData(true, false, false)]
[InlineData(false, true, true)]
[InlineData(true, false, true)]
[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 enforced, bool finalRetained, bool refresh2, bool resubscribe)
{
var ct = TestContext.Current.CancellationToken;
await using var rig = await ServerEventRig.StartAsync("Refresh854", encrypted: false, ct);
@@ -100,10 +107,21 @@ public sealed class ConditionRefreshDeferralWireTests
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"));
@@ -149,6 +167,144 @@ public sealed class ConditionRefreshDeferralWireTests
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
@@ -294,6 +450,7 @@ public sealed class ConditionRefreshDeferralWireTests
}
}, stop.Token, TaskCreationOptions.LongRunning, TaskScheduler.Default);
var refusedAsInProgress = 0;
for (var r = 0; r < 40; r++)
{
var call = r % 3 == 2
@@ -301,7 +458,17 @@ public sealed class ConditionRefreshDeferralWireTests
serverSubscription.Id, onServer.Status.Id)
: serverSession.CallAsync(ObjectTypeIds.ConditionType, MethodIds.ConditionType_ConditionRefresh, ct,
serverSubscription.Id);
await call.WaitAsync(TimeSpan.FromSeconds(20), ct);
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);
}
@@ -311,16 +478,17 @@ public sealed class ConditionRefreshDeferralWireTests
Volatile.Read(ref deferredSeen).ShouldBeGreaterThan(1, "the stress must actually have run with deferred events");
await RecordedEvents.WaitUntilAsync(() => rig.Nm.DeferredEventReports == 0);
var last = $"s{Volatile.Read(ref counter)}";
var newest = $"s{Volatile.Read(ref counter)}";
await folderSubscription.SetPublishingModeAsync(true, ct);
await RecordedEvents.WaitUntilAsync(() => serverEvents.Has(onServer, last));
await RecordedEvents.WaitUntilAsync(() => serverEvents.Has(onServer, newest));
await Task.Delay(TimeSpan.FromMilliseconds(500), ct);
// Walk the Server-object stream: after each RefreshEnd, a condition event must not be older than what that
// refresh replayed (the refresh replays the then-current state, so the floor is its replayed counter).
var floor = -1;
// 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 replayedInThisRefresh = -1;
var violations = new List<string>();
foreach (var f in serverEvents.Of(onServer))
{
@@ -328,33 +496,139 @@ public sealed class ConditionRefreshDeferralWireTests
if (type == ObjectTypeIds.RefreshStartEventType)
{
replaying = true;
replayedInThisRefresh = -1;
continue;
}
if (type == ObjectTypeIds.RefreshEndEventType)
{
replaying = false;
floor = Math.Max(floor, replayedInThisRefresh);
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)
{
replayedInThisRefresh = n;
continue;
}
if (n < floor) violations.Add($"s{n} after a refresh that replayed s{floor}");
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(last, "the stream still converges on the newest state");
await RecordedEvents.WaitUntilAsync(() => folderEvents.Has(onA, last));
.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>