fix(opcua): ConditionRefresh on a Server-object item is not overtaken by older deferred condition events (#854)
While a folder subscriber's SDK channel is full, a condition's later events wait in its #838 deferral queue and reach i=2253 items only when the drain gets to them, off Lock. A ConditionRefresh in that window replayed the current state, then the older deferred events followed RefreshEnd (stale state, or a condition the refresh had just dropped re-added until the newest drained). - Condition events are reported as ConditionEventSnapshot (an InstanceStateSnapshot carrying a report sequence, condition NodeId and Retain); events queued behind a deferred report are marked Deferred. - Server-object refresh (ConditionRefresh and ConditionRefresh2, flag on AND off) now runs in RefreshServerObjectItems: watermark, "anything deferred?", the replay and its queueing in ONE Lock hold, so the replay is exactly the state at the cut. Per item it queues the replay and installs a RefreshCut under the item's gate. - OnConditionReportEvent holds that gate across "superseded? -> QueueEvent" for deferred events only. Withheld from that item alone: events at/below the cut whose condition the replay re-sent, or Retain=true events of a condition it omitted; an older Retain=false event of an omitted condition is still delivered. Queues, other Server-object items and folder items are untouched; refresh never waits for a drain. - Lock order Lock -> item gate -> SDK item lock, same as a synchronous report. - Meter otopcua.opcua.condition_events.superseded. - Folder items keep a documented residual (SDK channel backlog + memoised per-item verdict leave no per-event hook). Tests: ConditionRefreshDeferralWireTests (deterministic overload by holding the folder item's SDK lock; 6 variants fail before the fix; probe-driven gate test fails with the gate removed; concurrent reports/refreshes/stall stress test fails before the fix).
This commit is contained in:
Joseph Doherty
@@ -211,8 +211,10 @@ 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) 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 —
|
||||
`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`,
|
||||
|
||||
+27
-4
@@ -882,15 +882,38 @@ 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). 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.
|
||||
- ⚠️ **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; 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.
|
||||
|
||||
**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>
|
||||
|
||||
@@ -1386,7 +1386,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 +1435,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 +1526,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
|
||||
@@ -1651,6 +1662,65 @@ 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>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)
|
||||
{
|
||||
/// <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).
|
||||
/// </summary>
|
||||
public bool Supersedes(ConditionEventSnapshot e) =>
|
||||
e.Sequence <= Watermark && (e.Retain || Refreshed.Contains(e.ConditionId));
|
||||
}
|
||||
|
||||
|
||||
/// <summary>
|
||||
/// The gate-relevant slice of a Part 9 condition's state — exactly the fields that drive a
|
||||
@@ -5292,7 +5362,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, byte> _allEventsItems = new();
|
||||
private readonly ConcurrentDictionary<IEventMonitoredItem, AllEventsItemState> _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
|
||||
@@ -5314,7 +5384,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3
|
||||
lock (Lock)
|
||||
{
|
||||
if (unsubscribe) _allEventsItems.TryRemove(monitoredItem, out _);
|
||||
else _allEventsItems[monitoredItem] = 0;
|
||||
else _allEventsItems.GetOrAdd(monitoredItem, static _ => new AllEventsItemState());
|
||||
ReconcileAllEventsMonitoring();
|
||||
}
|
||||
|
||||
@@ -5488,8 +5558,9 @@ 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)
|
||||
foreach (var (item, state) in _allEventsItems)
|
||||
{
|
||||
if (e is AuditEventState && !IsAuditDeliverableTo(item)) continue;
|
||||
if (scopes is not null && !IsConditionVisibleTo(item, scopes)) continue;
|
||||
@@ -5501,10 +5572,36 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3
|
||||
continue;
|
||||
}
|
||||
|
||||
item.QueueEvent(e);
|
||||
if (deferred is null)
|
||||
{
|
||||
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.
|
||||
lock (state.Gate)
|
||||
{
|
||||
if (state.Cut?.Supersedes(deferred) == true)
|
||||
{
|
||||
Interlocked.Increment(ref _refreshSupersededDeliveries);
|
||||
OtOpcUaTelemetry.ConditionEventDeliveriesSuperseded.Add(1);
|
||||
continue;
|
||||
}
|
||||
|
||||
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,19 +6165,15 @@ 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)
|
||||
@@ -6089,36 +6182,116 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3
|
||||
(item.NodeId == ObjectIds.Server ? serverItems : direct).Add(item);
|
||||
}
|
||||
|
||||
var enforced = AclEnforced;
|
||||
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 (direct.Count > 0)
|
||||
{
|
||||
if (ConditionOf(e) is not { } condition || !seen.Add(condition)) continue;
|
||||
conditions.Add((e, ConditionVisibilityScopes(condition)));
|
||||
if (enforced) RefreshDirectItems(systemContext, direct);
|
||||
else base.ConditionRefresh(context, direct);
|
||||
}
|
||||
|
||||
foreach (var item in serverItems)
|
||||
if (serverItems.Count > 0) RefreshServerObjectItems(systemContext, serverItems, enforced);
|
||||
return ServiceResult.Good;
|
||||
}
|
||||
|
||||
/// <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>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.
|
||||
/// </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).
|
||||
/// </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.
|
||||
/// </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.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
private void RefreshServerObjectItems(ServerSystemContext systemContext, List<IEventMonitoredItem> items, bool enforced)
|
||||
{
|
||||
lock (Lock)
|
||||
{
|
||||
foreach (var (e, scopes) in conditions)
|
||||
var watermark = Interlocked.Read(ref _conditionEventSequence);
|
||||
bool deferring;
|
||||
lock (_eventReportGate) deferring = _deferredEventReports > 0;
|
||||
|
||||
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 (!IsConditionVisibleTo(item, scopes)) continue;
|
||||
if (ServiceResult.IsBad(ValidateEventRolePermissions(item, e))) continue;
|
||||
item.QueueEvent(e);
|
||||
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 item in items)
|
||||
{
|
||||
var queued = new List<IFilterTarget>();
|
||||
var refreshed = new HashSet<NodeId>();
|
||||
foreach (var (e, condition, 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);
|
||||
}
|
||||
|
||||
if (!_allEventsItems.TryGetValue(item, out var state))
|
||||
{
|
||||
// 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;
|
||||
}
|
||||
|
||||
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;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return ServiceResult.Good;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
|
||||
+430
@@ -0,0 +1,430 @@
|
||||
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.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>
|
||||
[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)]
|
||||
public async Task Deferred_events_older_than_a_refresh_never_reach_the_refreshed_server_object_item_after_RefreshEnd(
|
||||
bool enforced, bool finalRetained, bool refresh2)
|
||||
{
|
||||
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");
|
||||
stall.Release();
|
||||
}
|
||||
|
||||
await RecordedEvents.WaitUntilAsync(() => rig.Nm.DeferredEventReports == 0);
|
||||
// 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>
|
||||
/// 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);
|
||||
|
||||
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);
|
||||
await call.WaitAsync(TimeSpan.FromSeconds(20), ct);
|
||||
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 last = $"s{Volatile.Read(ref counter)}";
|
||||
await folderSubscription.SetPublishingModeAsync(true, ct);
|
||||
await RecordedEvents.WaitUntilAsync(() => serverEvents.Has(onServer, last));
|
||||
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;
|
||||
var replaying = false;
|
||||
var replayedInThisRefresh = -1;
|
||||
var violations = new List<string>();
|
||||
foreach (var f in serverEvents.Of(onServer))
|
||||
{
|
||||
var type = f[RecordedEvents.EventTypeField].GetNodeId();
|
||||
if (type == ObjectTypeIds.RefreshStartEventType)
|
||||
{
|
||||
replaying = true;
|
||||
replayedInThisRefresh = -1;
|
||||
continue;
|
||||
}
|
||||
|
||||
if (type == ObjectTypeIds.RefreshEndEventType)
|
||||
{
|
||||
replaying = false;
|
||||
floor = Math.Max(floor, replayedInThisRefresh);
|
||||
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}");
|
||||
}
|
||||
|
||||
violations.ShouldBeEmpty();
|
||||
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));
|
||||
}
|
||||
|
||||
/// <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();
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user