Merge #855 folder event items: per-item delivery under the node manager, ConditionRefresh cut matches the Server object
Claude-Session: https://claude.ai/code/session_01GT67AD99zQQqpLCVm3p7xk
This commit is contained in:
@@ -217,7 +217,7 @@ Two things to know before relying on any of this:
|
||||
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 —
|
||||
`ConditionRefreshDeferralWireTests`. ⚠️ **Since #855 folder items are delivered per item too:** at every notifier we own, `DeliverConditionEventAsync` serves the `MonitoredNode2`'s attached items itself (`DeliverToNotifierItems`: audit guard → `INodeManager3.ValidateEventRolePermissions` per event, UNCACHED → item gate/cut → `QueueEvent`), never through the SDK channel — so no notifier of ours defers, a flood under `Lock` defers nothing, and folder refreshes (`RefreshDirectItems`, flag on AND off) share the #854 cut. The deferral queue + cut remain as the safety net; an event whose walk suspends is marked deferred BEFORE suspending. Tests produce deferral with `NotifierStall` (the `NotifierDeliveryStall` probe) — holding a folder item's SDK lock now blocks the reporter. `FolderConditionRefreshWireTests`; `docs/plans/2026-11-11-folder-event-per-item-delivery.md`. 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`. Overflow WITH the ACL doing work (#838
|
||||
item 2) — denied condition interleaved, folder + `i=2253` items, non-publishing client, concurrent Read, grant revoked
|
||||
|
||||
@@ -0,0 +1,102 @@
|
||||
# Folder (notifier) event items: per-item delivery (Gitea #855)
|
||||
|
||||
Status: design for #855 (owner decision 2026-11-11: build per-item delivery). Background: #838 (deferral instead of
|
||||
blocking under `Lock`), #854 (sequence cut on Server-object refresh), #826 (per-folder event ACL).
|
||||
|
||||
## Problem
|
||||
|
||||
A condition event reaches an event item on an equipment / raw device folder through the SDK's `MonitoredNode2`
|
||||
for that folder: `OnReportEventAsync` writes it to a bounded channel (4096), and a per-node consumer task decides
|
||||
each item's verdict (memoised per `(item, event type, SourceNode)`) and calls `QueueEvent`. That gives the node
|
||||
manager no per-event, per-item decision point, so a `ConditionRefresh` of a folder item cannot withhold an older
|
||||
event that is still in the channel or in our deferral queue: the refresh is overtaken (the #854 guarantee covers only
|
||||
Server-object items). The channel's consumer also needs `Lock` on a verdict miss, which is why a burst under `Lock`
|
||||
fills the channel and defers (#838).
|
||||
|
||||
## Design
|
||||
|
||||
### Where delivery hooks in
|
||||
|
||||
`DeliverConditionEventAsync` already walks the condition's notifier chain hop by hop (our copy of the SDK's
|
||||
`ReportEventAsync`, #838). At a hop whose `OnReportEventAsync` is the SDK `MonitoredNode2` of **a node this manager
|
||||
owns** (`handler.Target is MonitoredNode2 m && m.Node == node`), the walk no longer calls the handler. It delivers the
|
||||
event itself to that node's attached items (`m.EventMonitoredItems`), one by one: the SDK's audit guard, the verdict
|
||||
(`INodeManager3.ValidateEventRolePermissions` — the SDK's type/source `ReceiveEvents` check plus, under enforcement,
|
||||
the #826 live-notifier + grant check), then the item's gate and cut, then `QueueEvent`.
|
||||
|
||||
This is exactly how Server-object items are served (`OnConditionReportEvent`, #586/#854): same verdict per event, same
|
||||
gate, same cut, same `QueueEvent`.
|
||||
|
||||
### Bypass / coexistence with `MonitoredNode2`
|
||||
|
||||
- **Condition events bypass the channel** for our notifiers. The `MonitoredNode2` stays: the SDK still owns item
|
||||
attachment (subscribe / unsubscribe / transfer / tombstone / revalidation stand-down detach), and its channel still
|
||||
carries everything else that reaches the node — a condition's SDK-raised method **audit** events (only with auditing
|
||||
on) and data-change notifications for attribute items on the folder. Those keep the SDK's cached verdict, so
|
||||
`InvalidateFolderEventVerdicts` stays.
|
||||
- A stood-down item is not in `EventMonitoredItems`, so it receives nothing, as before.
|
||||
- A hop whose handler is anything else (none exists today: conditions are never event notifiers and variables cannot
|
||||
host event items) still goes through the handler — the fallback that keeps the #838 deferral reachable.
|
||||
|
||||
### Consequence for overload
|
||||
|
||||
Every hop the node manager owns now completes synchronously: `QueueEvent` never blocks (the item's own queue discards
|
||||
oldest at `QueueSize`), and the verdict runs on the reporter's thread, which already holds `Lock`. So a burst under one
|
||||
`Lock` hold no longer defers at all, and the #838 deadlock shape (a consumer needing `Lock` while the reporter waits on
|
||||
its channel) is gone for condition events. Folder items stay in report order per item, exactly like Server-object items.
|
||||
|
||||
The #838 deferral queue and the #854 cut machinery remain as the safety net for a hop that does suspend (the fallback
|
||||
above, or a future SDK change). Folder items are brought under the same cut:
|
||||
|
||||
- An event whose delivery continues off `Lock` — queued in the deferral queue, **or** the in-flight event whose walk
|
||||
suspended at a hop (new: marked when the walk suspends, so the hops after the suspension take the gate) — takes the
|
||||
item's gate and is withheld when the item's latest refresh cut supersedes it (`RefreshCut.Supersedes`, unchanged).
|
||||
- `ConditionRefresh` / `ConditionRefresh2` of folder items has the same two paths as the Server object, flag on and off:
|
||||
- **Nothing deferred:** the SDK base walk of the item's notifier (off `Lock`, live condition nodes), per-item verdict
|
||||
as before (#826 filter under enforcement). Kept only if no deferral episode existed at any moment of it; otherwise
|
||||
the cut path runs too and supersedes it inside the same RefreshStart..RefreshEnd. Its closing check takes `Lock`
|
||||
once: a report whose walk suspended (marked under `Lock` before suspending) has registered its deferral by then —
|
||||
without it a folder refresh could pass the check in the instant between the suspension and the count.
|
||||
- **Deferring — the cut:** verdicts per (item, condition) first, off `Lock`, from a live walk. Then one `Lock` hold:
|
||||
watermark, episode, the Retain set, and per item the same walk producing frozen snapshots, with the cut + "replay
|
||||
pending" marker installed under the item's gate. The replay is queued off `Lock`, one event per gate hold.
|
||||
- State per item (`EventItemState`, previously Server-object only) is keyed by the item object for its lifetime, so it
|
||||
survives unsubscribe + resubscribe; stale cuts are cleared when the deferral episode ends, for folder and Server items.
|
||||
|
||||
### Locking
|
||||
|
||||
Unchanged order, now for folder items too: `Lock` → item gate → the SDK item's own lock. Synchronous delivery runs
|
||||
under the reporter's `Lock`; the verdict takes `Lock` re-entrantly and the core node manager's / authorizer's locks
|
||||
(as the Server-object path already does). Off-`Lock` (deferred) delivery computes the verdict **before** taking the item
|
||||
gate, so nothing under a gate takes `Lock` or the event-report gate. A reporter waits at most for one in-progress
|
||||
`QueueEvent` on an item; a refresh waits at most for one deferred delivery already inside the gate, never for a drain.
|
||||
No reporter ever waits on a channel.
|
||||
|
||||
### Bounds
|
||||
|
||||
No new queue. Per item: the SDK item queue (`QueueSize`, DiscardOldest). Deferral (only on the fallback): unchanged,
|
||||
1024 per condition / 65 536 total, coalescing. Cuts: one per refreshed item, sets sized by the conditions, cleared at the
|
||||
end of the deferral episode.
|
||||
|
||||
### ACL (#826)
|
||||
|
||||
Each event is decided per item at delivery against the current generation — no memoised verdict for condition events
|
||||
any more, so a deploy / revocation / reference removal takes effect on the next event without waiting for an
|
||||
invalidation. Refresh replays only conditions the item may see (the same check); a condition omitted for want of a grant
|
||||
is never withheld afterwards (the cut's existing rule).
|
||||
|
||||
### Cost
|
||||
|
||||
The per-event verdict for folder items moves from the SDK's consumer task (cached) to the reporter thread under `Lock`
|
||||
(uncached), as Server-object items already pay. Folder subscribers are few (ScadaBridge subscribes at `i=2253`).
|
||||
|
||||
## Tests
|
||||
|
||||
- `ConditionRefreshDeferralWireTests` / `FolderEventFloodWireTests` / `EventQueueOverflowAclWireTests`: deferral is no
|
||||
longer produced by flooding a folder item; tests that need it stall the folder's hop through a test probe
|
||||
(`NotifierDeliveryStall`, null in production) instead of the folder item's SDK lock (which would now block the reporter,
|
||||
as it would on the Server object). Flood tests now pin that folder delivery never defers.
|
||||
- New `FolderConditionRefreshWireTests`: folder-item refresh during deferral (flag on/off, retained / cleared,
|
||||
ConditionRefresh2, resubscribe), the gate race, newer-than-cut replay, ACL omission not withheld, ACL revocation during
|
||||
deferral, the in-flight suspended event, and two stress tests (production overload with no stall; stall + refresh +
|
||||
reports concurrently).
|
||||
+46
-22
@@ -901,7 +901,7 @@ Behaviours worth knowing before enabling it:
|
||||
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
|
||||
- The deferral queues, every other Server-object item and every non-refreshing 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.
|
||||
@@ -912,29 +912,53 @@ Behaviours worth knowing before enabling it:
|
||||
- **Measured** (3000 retained conditions, longest `Lock` wait another thread saw during one refresh): the first
|
||||
#854 cut (`1622d07f`, everything under `Lock`) 95–106 ms idle / 102–127 ms deferring; now 0–3 ms idle /
|
||||
23–32 ms deferring (the frozen snapshots).
|
||||
- ⚠️ **Residual: folder (notifier) items.** A refresh of an item on an equipment/device folder can still be followed
|
||||
by older events of a condition. Two sources, neither reachable per item: the SDK's own `MonitoredNode2` channel
|
||||
(up to 4096 events already accepted before the refresh — this exists in the stock SDK too, and in normal operation
|
||||
for the few events in transit), and our deferral queue, whose folder hop goes through that channel, where the SDK
|
||||
memoises the per-item permission verdict by (item, event type, source) so there is no per-event, per-item hook. The
|
||||
events stay in the condition's order and its newest arrives last, so the client converges. Discarding the queue
|
||||
globally on refresh would drop transitions for every non-refreshing subscriber while leaving the channel backlog;
|
||||
making refresh wait would block a service call on the stalled drain. ScadaBridge also ignores a transition older
|
||||
than the one it holds for a condition.
|
||||
- **Folder (notifier) items are delivered per item (Gitea #855).** Before #855 a folder item was fed by the SDK's own
|
||||
`MonitoredNode2` channel (up to 4096 events accepted before a refresh), with the per-item verdict memoised by (item,
|
||||
event type, source), so a refresh of a folder item could still be followed by older events. Now, at every notifier
|
||||
the node manager owns, the walk serves the attached items itself (`DeliverToNotifierItems`), exactly as it serves
|
||||
Server-object items: the SDK's audit guard, the verdict through `INodeManager3.ValidateEventRolePermissions` (SDK
|
||||
type/source `ReceiveEvents` + the #826 decision under enforcement — its item-independent part, the live condition,
|
||||
the notifier check and the visibility scopes, resolved once per hop) **per event, uncached**, then the item's gate and
|
||||
cut, then `QueueEvent`. The `MonitoredNode2` stays for item attachment and for what still uses its channel (a
|
||||
condition's SDK method audit events, data-change items on the folder), so `InvalidateFolderEventVerdicts` stays.
|
||||
- **No notifier of ours defers any more.** `QueueEvent` never waits for a consumer (the item queue discards oldest at
|
||||
`QueueSize`) and the verdict runs on the reporter's thread, which already holds `Lock` — so a burst under one
|
||||
`Lock` hold completes with nothing deferred, and the #838 deadlock shape is gone for condition events. Folder items
|
||||
therefore see condition events in report order. The cost moves with it: the per-event verdict and filter for
|
||||
folder items now run under `Lock` (as for Server-object items), where the SDK used to cache the verdict off it —
|
||||
including the SDK item queue's own per-event costs (its duplicate scan, and at a full queue a list shift), so many
|
||||
folder items on one notifier whose clients stop publishing make each report proportionally slower under `Lock`.
|
||||
With auditing on (off in this repo), a condition's SDK method **audit** event still takes the folder's channel, so
|
||||
at a folder item it may now arrive after the condition event that method produced (they were one FIFO before).
|
||||
- **The deferral queue and the cut stay as the safety net**, for a hop that is not ours (none exists today:
|
||||
conditions are never event notifiers and variables cannot host event items) — and folder items share the cut:
|
||||
a `ConditionRefresh`/`ConditionRefresh2` of a folder item takes the same two paths as the Server object (plain walk
|
||||
off `Lock` when nothing is deferred; otherwise verdicts off `Lock`, then one `Lock` hold for the watermark, the
|
||||
Retain set and, per item, its notifier walk frozen as snapshots, with the cut installed under the item gate and the
|
||||
replay queued off `Lock`). An event whose walk suspended at a hop is marked deferred **before** it suspends, so
|
||||
every later hop judges it against the cut; the plain refresh path's closing check takes `Lock` once, so a report
|
||||
that suspended under it has registered its deferral by then. Lock order is unchanged (`Lock` → item gate → item
|
||||
lock); off `Lock` the verdict is computed before the gate.
|
||||
|
||||
Pinned by `FolderEventFloodWireTests`: 6000 events under one `Lock` hold with publishing disabled (flag on and off)
|
||||
complete, drain to zero, coalesce, and the item delivers at most its revised `QueueSize` in increasing order ending
|
||||
with the last event; and a condition **removed** in the same `Lock` hold has none of its deferred events delivered to
|
||||
either the folder or the Server-object item (the flag-off variant fails without the liveness checks).
|
||||
`ConditionRefreshDeferralWireTests` (#854) stalls a folder item's SDK lock so its channel fills without `Lock` held,
|
||||
refreshes the Server-object item (ConditionRefresh and ConditionRefresh2, flag on/off, condition retained or cleared)
|
||||
and pins that nothing older than the replay follows RefreshEnd while the non-refreshing folder item still receives the
|
||||
whole tail (also across an unsubscribe + resubscribe of the item, and the cut is cleared after the drain); a
|
||||
probe-driven test pins that a refresh waits for a delivery already past its check (fails without the item gate); an
|
||||
ACL-on test pins that a condition omitted for want of a grant is not withheld once the grant lands; a 3000-condition
|
||||
test measures the `Lock` hold and pins that neither path queues its replay under `Lock`; and a stress test runs
|
||||
reports, refreshes and an intermittent stall concurrently (its stream never goes backwards outside a refresh, and
|
||||
every refresh ends on the newest state delivered).
|
||||
complete **with nothing deferred** (fails if the per-item delivery stops recognising the SDK fan-out), and the item
|
||||
delivers at most its revised `QueueSize` in increasing order ending with the last event; and a condition **removed**
|
||||
while its events wait (delivery stalled at the folder) has none of them delivered to either the folder or the
|
||||
Server-object item. Tests that need a deferral stall delivery at a folder through a test probe
|
||||
(`NotifierStall` → `OtOpcUaNodeManager.NotifierDeliveryStall`, null in production), which suspends the walk exactly
|
||||
as a full channel did — holding a folder item's SDK lock, as the pre-#855 tests did, would now block the reporter.
|
||||
`ConditionRefreshDeferralWireTests` (#854) refreshes the Server-object item while the folder is stalled
|
||||
(ConditionRefresh and ConditionRefresh2, flag on/off, condition retained or cleared) and pins that nothing older than
|
||||
the replay follows RefreshEnd while the non-refreshing folder item still receives the whole tail (also across an
|
||||
unsubscribe + resubscribe of the item, and the cut is cleared after the drain); a probe-driven test pins that a
|
||||
refresh waits for a delivery already past its check; 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 a stress test runs reports,
|
||||
refreshes and an intermittent stall concurrently. `FolderConditionRefreshWireTests` (#855) is the folder-item twin:
|
||||
the same matrix on a refreshed folder item (a second subscriber on the same folder keeps the whole tail; a
|
||||
ModifyMonitoredItems re-subscribe keeps the cut), the gate race, a newer-than-cut event on the raw device folder, a
|
||||
condition omitted by the ACL (reference dropped) not withheld once it returns, a revocation after the refresh, a
|
||||
production-shape overload (bursts under `Lock` + repeated refreshes, no stall: nothing deferred, the stream never
|
||||
goes backwards, no replay queued under `Lock`) and a stress test with an intermittent stall.
|
||||
|
||||
**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
|
||||
|
||||
@@ -1476,7 +1476,14 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3
|
||||
/// synchronous: the walk completes inline while every channel has room.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// <b>Deferral.</b> When a channel is full the walk continues asynchronously (holding no thread or lock),
|
||||
/// <b>Since Gitea #855 no notifier of ours is a channel.</b> At a notifier this manager owns, the walk serves
|
||||
/// the items itself (<see cref="DeliverToNotifierItems"/>) instead of writing to the SDK's
|
||||
/// <c>MonitoredNode2</c> channel, so it always completes inline and the deferral below is reached only through
|
||||
/// a hop that is not ours (none exists today) — or the test stall (<see cref="NotifierDeliveryStall"/>). It is
|
||||
/// kept as the safety net, and the refresh cut (#854) covers folder items too.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// <b>Deferral.</b> When a hop suspends the walk continues asynchronously (holding no thread or lock),
|
||||
/// and the condition gets a queue: its later events wait behind it, so ONE condition's events stay in
|
||||
/// order, while other conditions whose channels have room are not held up (per-condition, not global).
|
||||
/// Events of different conditions may interleave differently than reported — each carries its own
|
||||
@@ -1522,13 +1529,29 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3
|
||||
|
||||
if (delivery.IsCompletedSuccessfully) return;
|
||||
|
||||
// #855 review: a delivery that FAILED inline is over, not suspended. Booking it as a deferral used to open and close
|
||||
// a deferral episode synchronously (moving the episode a concurrent plain refresh checks, and clearing cuts under
|
||||
// the event-report gate while Lock was held). Only an incomplete delivery is deferred.
|
||||
if (delivery.IsCompleted)
|
||||
{
|
||||
_logger.LogError(delivery.AsTask().Exception?.GetBaseException(),
|
||||
"OtOpcUaNodeManager: failed to report Part 9 condition event for {NodeId}", condition.NodeId);
|
||||
return;
|
||||
}
|
||||
|
||||
DeferredConditionReports queue;
|
||||
lock (_eventReportGate)
|
||||
{
|
||||
var queue = new DeferredConditionReports();
|
||||
queue = new DeferredConditionReports();
|
||||
_deferredReports[condition] = queue;
|
||||
CountDeferred(+1, condition);
|
||||
_ = DrainDeferredAsync(condition, queue, delivery.AsTask());
|
||||
}
|
||||
|
||||
// #855 review 2: started on the pool, never inline. The suspended delivery may complete between the check above and
|
||||
// here; an inline drain would then run (and end the episode, and clear cuts over every item gate) on this thread,
|
||||
// under Lock. The queue is registered first, so this condition's later reports wait behind it either way.
|
||||
var inFlight = delivery.AsTask();
|
||||
_ = Task.Run(() => DrainDeferredAsync(condition, queue, inFlight));
|
||||
}
|
||||
|
||||
/// <summary>Queues a later event behind a condition's deferred one, coalescing past the budget. Under the gate.</summary>
|
||||
@@ -1615,7 +1638,42 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3
|
||||
|
||||
node.OnReportEvent?.Invoke(SystemContext, node, e);
|
||||
if (node.OnReportEventAsync is { } handler)
|
||||
await handler(SystemContext, node, e, CancellationToken.None).ConfigureAwait(false);
|
||||
{
|
||||
if (OwnNotifierFanOut(node, handler) is { } monitored)
|
||||
{
|
||||
// Gitea #855: one of our notifiers. Its items are served here, per item and per event, exactly as the
|
||||
// Server-object items are (OnConditionReportEvent) — never through the SDK's channel, whose consumer
|
||||
// needed Lock (#838) and gave no per-event, per-item point at which a refresh cut could apply.
|
||||
if (NotifierDeliveryStall?.Invoke(node) is { IsCompleted: false } stall)
|
||||
{
|
||||
MarkDeferred(e);
|
||||
await stall.ConfigureAwait(false);
|
||||
}
|
||||
|
||||
// Re-checked: a stall may have outlived the condition (the liveness check above ran before it).
|
||||
if (IsLiveCondition(condition))
|
||||
{
|
||||
// #855 review 3: one hop's failure must not cost the hops after it (before #855 this hop was a channel
|
||||
// write, which cannot throw; the SDK isolated failures per item).
|
||||
try
|
||||
{
|
||||
DeliverToNotifierItems(monitored, e);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogWarning(ex, "OtOpcUaNodeManager: failed to deliver a condition event at notifier {NodeId}", node.NodeId);
|
||||
}
|
||||
}
|
||||
}
|
||||
else
|
||||
{
|
||||
// Not a fan-out of ours (none exists today): the SDK path, which may suspend. Whatever is delivered after
|
||||
// the suspension runs off Lock, so it must be judged as a deferred event (#855).
|
||||
var pending = handler(SystemContext, node, e, CancellationToken.None);
|
||||
if (!pending.IsCompleted) MarkDeferred(e);
|
||||
await pending.ConfigureAwait(false);
|
||||
}
|
||||
}
|
||||
|
||||
var notifiers = new List<NodeState.Notifier>();
|
||||
node.GetNotifiers(SystemContext, notifiers);
|
||||
@@ -1631,6 +1689,101 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3
|
||||
await DeliverConditionEventAsync(condition, parent, e).ConfigureAwait(false);
|
||||
}
|
||||
|
||||
/// <summary>Marks an event whose remaining delivery will run off <c>Lock</c> (#855) — before the walk suspends.</summary>
|
||||
private static void MarkDeferred(IFilterTarget e)
|
||||
{
|
||||
if (e is ConditionEventSnapshot snapshot) snapshot.Deferred = true;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// The SDK <see cref="MonitoredNode2"/> serving event items on <paramref name="node"/>, when that node is one of
|
||||
/// ours and <paramref name="handler"/> is the fan-out's own <c>OnReportEventAsync</c> (what
|
||||
/// <c>MonitoredNode2.Add(IEventMonitoredItem)</c> installs); otherwise null (Gitea #855).
|
||||
/// </summary>
|
||||
private MonitoredNode2? OwnNotifierFanOut(NodeState node, NodeStateReportEventAsyncHandler handler) =>
|
||||
handler.Target is MonitoredNode2 monitored && ReferenceEquals(monitored.Node, node) && OwnsNamespace(node.NodeId)
|
||||
? monitored
|
||||
: null;
|
||||
|
||||
/// <summary>
|
||||
/// Test probe (#855): consulted before a condition event is delivered to the items of one of our notifiers; an
|
||||
/// incomplete task is awaited first, so the walk suspends there exactly as it would behind a slow notifier — the
|
||||
/// way tests produce a deferral now that no notifier of ours can block. Null in production.
|
||||
/// </summary>
|
||||
internal Func<NodeState, Task?>? NotifierDeliveryStall { get; set; }
|
||||
|
||||
/// <summary>
|
||||
/// Delivers one condition event to the event items attached to one of this manager's notifiers (an equipment or
|
||||
/// raw device folder), item by item (Gitea #855) — the folder twin of <see cref="OnConditionReportEvent"/>.
|
||||
/// </summary>
|
||||
/// <param name="monitored">The notifier's SDK fan-out; only its attached-item list is used.</param>
|
||||
/// <param name="e">The condition's snapshot.</param>
|
||||
/// <remarks>
|
||||
/// <para>
|
||||
/// Mirrors the SDK's <c>MonitoredNode2.ProcessEventForItemAsync</c> — its audit guard, then the verdict through
|
||||
/// <see cref="INodeManager3.ValidateEventRolePermissions"/> (the SDK's type + SourceNode <c>ReceiveEvents</c>
|
||||
/// check and, under enforcement, #826's live-notifier + grant check), then <c>QueueEvent</c> — but per event:
|
||||
/// nothing is memoised, so a deploy, revocation or reference removal decides the very next event. The item's
|
||||
/// gate and refresh cut apply exactly as on the Server object (<see cref="QueueConditionEventToItem"/>).
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// Synchronous. On a report it runs on the reporter's thread under <c>Lock</c> (as the Server-object delivery
|
||||
/// does), and <c>QueueEvent</c> never waits for a consumer — the item's own queue discards oldest at its
|
||||
/// <c>QueueSize</c> — so no notifier of ours defers any more. Off <c>Lock</c> (a deferred event) the verdict,
|
||||
/// which takes <c>Lock</c>, is computed before the item gate is taken: nothing under a gate takes <c>Lock</c>.
|
||||
/// A stood-down item (#562) is not in the attached list and receives nothing, as before.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
private void DeliverToNotifierItems(MonitoredNode2 monitored, IFilterTarget e)
|
||||
{
|
||||
var deferred = e is ConditionEventSnapshot { Deferred: true } d ? d : null;
|
||||
|
||||
// #855 review 2: the #826 decision's item-independent part — the live condition, whether this notifier is still a
|
||||
// legitimate notifier of it, and its visibility scopes — is resolved ONCE per hop, not per item (it is what
|
||||
// IsNotifierEventVisibleTo resolves for an item attached here). Only the grant check and the SDK's role check
|
||||
// remain per item.
|
||||
var enforced = AclEnforced;
|
||||
List<NodeId>? scopes = null;
|
||||
if (enforced)
|
||||
{
|
||||
if (EventConditionOf(e) is not { } named
|
||||
|| !_alarmConditions.TryGetValue(MapKey(named.NodeId), out var live)
|
||||
|| !IsLiveNotifierOf(live, monitored.Node))
|
||||
{
|
||||
return; // fail closed for every item on this notifier
|
||||
}
|
||||
|
||||
scopes = ConditionVisibilityScopes(live);
|
||||
}
|
||||
|
||||
foreach (var item in monitored.EventMonitoredItems.Values)
|
||||
{
|
||||
if (item is null) continue;
|
||||
try
|
||||
{
|
||||
if (IsAuditEvent(e) && !IsAuditDeliverableTo(item)) continue;
|
||||
if (ServiceResult.IsBad(ValidateEventRolePermissions(item, e))) continue;
|
||||
if (scopes is not null && !IsConditionVisibleTo(item, scopes)) continue;
|
||||
QueueConditionEventToItem(item, e, deferred, DeferredFolderDeliveryProbe);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
// As the SDK's per-item delivery: one item's failure must not cost the others (or the rest of the walk).
|
||||
_logger.LogWarning(ex, "OtOpcUaNodeManager: failed to deliver a condition event to event item {ItemId}", item.Id);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>The SDK's audit test from <c>MonitoredNode2.ProcessEventForItemAsync</c>: the event, or the snapshot's handle.</summary>
|
||||
private static bool IsAuditEvent(IFilterTarget e) =>
|
||||
e is AuditEventState || e is InstanceStateSnapshot { Handle: AuditEventState };
|
||||
|
||||
/// <summary>
|
||||
/// Test probe (#855): the folder twin of <see cref="DeferredServerObjectDeliveryProbe"/> — runs between a deferred
|
||||
/// event's supersession check and its queueing to a folder item, inside the item's gate. Null in production.
|
||||
/// </summary>
|
||||
internal Action? DeferredFolderDeliveryProbe { get; set; }
|
||||
|
||||
/// <summary>Whether <paramref name="condition"/> is the instance the address space currently holds for its NodeId.</summary>
|
||||
private bool IsLiveCondition(NodeState condition) =>
|
||||
_alarmConditions.TryGetValue(MapKey(condition.NodeId), out var live) && ReferenceEquals(live, condition);
|
||||
@@ -1653,9 +1806,9 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3
|
||||
{
|
||||
_lastDeferralWarningUtc = now;
|
||||
_logger.LogWarning(
|
||||
"OtOpcUaNodeManager: condition events are being deferred — an event notifier's delivery channel is "
|
||||
+ "full (first: {NodeId}). Folder event subscribers are not keeping up, or Lock is held across a large "
|
||||
+ "burst; events stay ordered per condition, and past {PerCondition} waiting per condition the oldest "
|
||||
"OtOpcUaNodeManager: condition events are being deferred — a notifier hop the node manager does not "
|
||||
+ "serve itself suspended (first: {NodeId}); since #855 no folder of ours does, so this is unexpected. "
|
||||
+ "Events stay ordered per condition, and past {PerCondition} waiting per condition the oldest "
|
||||
+ "give way to the newest (Gitea #838)",
|
||||
condition.NodeId, MaxDeferredEventsPerCondition);
|
||||
}
|
||||
@@ -1711,18 +1864,27 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3
|
||||
|
||||
public bool Retain { get; } = retain;
|
||||
|
||||
private volatile bool _deferred;
|
||||
|
||||
/// <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.
|
||||
/// Set when the event's delivery continues off <c>Lock</c>: under the event-report gate, before the drain can
|
||||
/// dequeue it, when it waits in a deferral queue; or by the walk itself, before it suspends at a hop (#855), for
|
||||
/// the in-flight event whose remaining hops then run on another thread. Only such an event reaches an item off
|
||||
/// <c>Lock</c>, which is the only way a refresh can be overtaken.
|
||||
/// </summary>
|
||||
public bool Deferred { get; set; }
|
||||
public bool Deferred
|
||||
{
|
||||
get => _deferred;
|
||||
set => _deferred = value;
|
||||
}
|
||||
}
|
||||
|
||||
/// <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.
|
||||
/// Per event item state — Server-object (#854) and, since #855, folder items: 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
|
||||
private sealed class EventItemState
|
||||
{
|
||||
/// <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>
|
||||
@@ -1749,21 +1911,21 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3
|
||||
private int _refreshGeneration;
|
||||
|
||||
/// <summary>
|
||||
/// One <see cref="AllEventsItemState"/> per Server-object item for the item's whole lifetime (#854 review): keyed by
|
||||
/// One <see cref="EventItemState"/> per event 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();
|
||||
private readonly ConditionalWeakTable<IEventMonitoredItem, EventItemState> _eventItemStates = new();
|
||||
|
||||
/// <summary>Server-object items currently holding a refresh cut (test observation, #854 review).</summary>
|
||||
/// <summary>Server-object and folder event items currently holding a refresh cut (test observation, #854 review).</summary>
|
||||
internal int ActiveRefreshCuts
|
||||
{
|
||||
get
|
||||
{
|
||||
var n = 0;
|
||||
foreach (var item in _allEventsItems.Keys)
|
||||
foreach (var item in TrackedEventItems())
|
||||
{
|
||||
if (!_allEventsItemStates.TryGetValue(item, out var state)) continue;
|
||||
if (!_eventItemStates.TryGetValue(item, out var state)) continue;
|
||||
lock (state.Gate) n += state.Cut is null ? 0 : 1;
|
||||
}
|
||||
|
||||
@@ -1771,8 +1933,15 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3
|
||||
}
|
||||
}
|
||||
|
||||
private AllEventsItemState AllEventsItemStateOf(IEventMonitoredItem item) =>
|
||||
_allEventsItemStates.GetValue(item, static _ => new AllEventsItemState());
|
||||
private EventItemState EventItemStateOf(IEventMonitoredItem item) =>
|
||||
_eventItemStates.GetValue(item, static _ => new EventItemState());
|
||||
|
||||
/// <summary>
|
||||
/// Every live event item this manager tracks: the Server-object items and the direct (folder) items (#855). An
|
||||
/// item that left both maps keeps its state in <see cref="_eventItemStates"/> until collected, but any cut it holds
|
||||
/// is from an ended episode by the time it could deliver again, and so inert.
|
||||
/// </summary>
|
||||
private IEnumerable<IEventMonitoredItem> TrackedEventItems() => _allEventsItems.Keys.Concat(_eventItems.Keys);
|
||||
|
||||
/// <summary>
|
||||
/// Advanced (under the event-report gate) each time the deferral count returns to zero: the end of a deferral
|
||||
@@ -1780,6 +1949,9 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3
|
||||
/// </summary>
|
||||
private long _deferralEpisode;
|
||||
|
||||
/// <summary>Deferral episodes ended since start (test observation, #855 review).</summary>
|
||||
internal long DeferralEpisodes => Interlocked.Read(ref _deferralEpisode);
|
||||
|
||||
/// <summary>What a Server-object item's latest ConditionRefresh conveyed (#854).</summary>
|
||||
/// <param name="Watermark">Every condition event with a sequence at or below this was reported before the replay was read.</param>
|
||||
/// <param name="Episode">The deferral episode the cut was taken in; once it ends the cut is stale and inert.</param>
|
||||
@@ -1810,9 +1982,9 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3
|
||||
private void ClearStaleRefreshCuts()
|
||||
{
|
||||
var current = Interlocked.Read(ref _deferralEpisode);
|
||||
foreach (var item in _allEventsItems.Keys)
|
||||
foreach (var item in TrackedEventItems())
|
||||
{
|
||||
if (!_allEventsItemStates.TryGetValue(item, out var state)) continue;
|
||||
if (!_eventItemStates.TryGetValue(item, out var state)) continue;
|
||||
lock (state.Gate)
|
||||
{
|
||||
if (state.Cut is { } cut && cut.Episode < current) state.Cut = null; // episodes only grow: never clear a newer cut (#854 re-review)
|
||||
@@ -5655,54 +5827,87 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3
|
||||
// through its OLD equipment folder (#808).
|
||||
if (!IsLiveCondition(node)) return;
|
||||
|
||||
var scopes = AclEnforced ? ConditionVisibilityScopes(node) : null;
|
||||
List<NodeId>? scopes;
|
||||
try
|
||||
{
|
||||
scopes = AclEnforced ? ConditionVisibilityScopes(node) : null;
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
// #855 review 3: withheld from the Server-object items (fail closed), but the walk goes on to the folders.
|
||||
_logger.LogWarning(ex, "OtOpcUaNodeManager: failed to decide a condition event for Server-object items ({NodeId})", node.NodeId);
|
||||
return;
|
||||
}
|
||||
|
||||
var reportingSession = (context as ISessionSystemContext)?.SessionId;
|
||||
var deferred = e is ConditionEventSnapshot { Deferred: true } d ? d : null;
|
||||
|
||||
foreach (var item in _allEventsItems.Keys)
|
||||
{
|
||||
if (e is AuditEventState && !IsAuditDeliverableTo(item)) continue;
|
||||
if (scopes is not null && !IsConditionVisibleTo(item, scopes)) continue;
|
||||
if (ServiceResult.IsBad(ValidateEventRolePermissions(item, e))) continue;
|
||||
if (reportingSession is { IsNull: false } reportingId
|
||||
&& item.Session?.Id is { } own
|
||||
&& !own.Equals(reportingId))
|
||||
try
|
||||
{
|
||||
continue;
|
||||
}
|
||||
|
||||
// #854: a synchronous event (under Lock) needs the gate only while a refresh's replay is pending, and the
|
||||
// pending marker is installed under Lock, so an item no refresh has touched has nothing to order. A deferred
|
||||
// event always takes the gate — its state is created here if need be, so a refresh racing it shares it.
|
||||
AllEventsItemState? state;
|
||||
if (deferred is not null) state = AllEventsItemStateOf(item);
|
||||
else if (!_allEventsItemStates.TryGetValue(item, out state) || state.PendingReplay == 0 || e is not ConditionEventSnapshot)
|
||||
{
|
||||
item.QueueEvent(e);
|
||||
continue;
|
||||
}
|
||||
|
||||
// A deferred event reaches this item off Lock, so a ConditionRefresh of the item may have replayed a newer
|
||||
// state since it was reported; and while a refresh's frozen replay is being queued, an event newer than it must
|
||||
// not be followed by it. Both are decided and queued under the item's gate, which the refresh also holds for
|
||||
// each replayed event it queues — so a deferred event lands before the replay or is judged against the cut.
|
||||
lock (state.Gate)
|
||||
{
|
||||
// A cut from an episode that has ended is stale (only waiting for ClearStaleRefreshCuts).
|
||||
if (deferred is not null
|
||||
&& state.Cut is { } cut && cut.Episode == Interlocked.Read(ref _deferralEpisode) && cut.Supersedes(deferred))
|
||||
if (e is AuditEventState && !IsAuditDeliverableTo(item)) continue;
|
||||
if (scopes is not null && !IsConditionVisibleTo(item, scopes)) continue;
|
||||
if (ServiceResult.IsBad(ValidateEventRolePermissions(item, e))) continue;
|
||||
if (reportingSession is { IsNull: false } reportingId
|
||||
&& item.Session?.Id is { } own
|
||||
&& !own.Equals(reportingId))
|
||||
{
|
||||
Interlocked.Increment(ref _refreshSupersededDeliveries);
|
||||
OtOpcUaTelemetry.ConditionEventDeliveriesSuperseded.Add(1);
|
||||
continue;
|
||||
}
|
||||
|
||||
if (state.PendingReplay != 0 && e is ConditionEventSnapshot { } ce && ce.Sequence > state.PendingWatermark)
|
||||
state.NewerSinceCut.Add(ce.ConditionId);
|
||||
|
||||
if (deferred is not null) DeferredServerObjectDeliveryProbe?.Invoke();
|
||||
item.QueueEvent(e);
|
||||
QueueConditionEventToItem(item, e, deferred, DeferredServerObjectDeliveryProbe);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
// #855 review: one item's failure (an item disposed under us, the SDK's operation guard at shutdown) must
|
||||
// not cost the other Server-object items, nor abort the walk before the folder hops.
|
||||
_logger.LogWarning(ex, "OtOpcUaNodeManager: failed to deliver a condition event to Server-object item {ItemId}", item.Id);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Queues a condition event to one event item — Server-object or folder (#855) — under the item's refresh gate and
|
||||
/// cut (#854). The caller has already decided the item may receive it, and holds no item gate.
|
||||
/// </summary>
|
||||
/// <param name="item">The item.</param>
|
||||
/// <param name="e">The event.</param>
|
||||
/// <param name="deferred">The event, when its delivery runs off <c>Lock</c> (<see cref="ConditionEventSnapshot.Deferred"/>).</param>
|
||||
/// <param name="probe">Test probe run inside the gate, before queueing a deferred event.</param>
|
||||
private void QueueConditionEventToItem(IEventMonitoredItem item, IFilterTarget e, ConditionEventSnapshot? deferred, Action? probe)
|
||||
{
|
||||
// #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.
|
||||
EventItemState? state;
|
||||
if (deferred is not null) state = EventItemStateOf(item);
|
||||
else if (!_eventItemStates.TryGetValue(item, out state) || state.PendingReplay == 0 || e is not ConditionEventSnapshot)
|
||||
{
|
||||
item.QueueEvent(e);
|
||||
return;
|
||||
}
|
||||
|
||||
// A deferred event reaches this item off Lock, so a ConditionRefresh of the item may have replayed a newer
|
||||
// state since it was reported; and while a refresh's frozen replay is being queued, an event newer than it must
|
||||
// not be followed by it. Both are decided and queued under the item's gate, which the refresh also holds for
|
||||
// each replayed event it queues — so a deferred event lands before the replay or is judged against the cut.
|
||||
lock (state.Gate)
|
||||
{
|
||||
// A cut from an episode that has ended is stale (only waiting for ClearStaleRefreshCuts).
|
||||
if (deferred is not null
|
||||
&& state.Cut is { } cut && cut.Episode == Interlocked.Read(ref _deferralEpisode) && cut.Supersedes(deferred))
|
||||
{
|
||||
Interlocked.Increment(ref _refreshSupersededDeliveries);
|
||||
OtOpcUaTelemetry.ConditionEventDeliveriesSuperseded.Add(1);
|
||||
return;
|
||||
}
|
||||
|
||||
if (state.PendingReplay != 0 && e is ConditionEventSnapshot { } ce && ce.Sequence > state.PendingWatermark)
|
||||
state.NewerSinceCut.Add(ce.ConditionId);
|
||||
|
||||
if (deferred is not null) probe?.Invoke();
|
||||
item.QueueEvent(e);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -6080,8 +6285,11 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3
|
||||
/// would be.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// Runs on the SDK's per-node delivery task, where the inherited role-permission check already takes
|
||||
/// <c>Lock</c> (via <c>GetNodeMetadata</c>); the scope resolution here takes it the same way.
|
||||
/// Runs per event, uncached, from the per-item folder delivery (#855) — on the reporter's thread under
|
||||
/// <c>Lock</c>, or off it for a deferred event (before the item gate is taken) — and, for the events that still
|
||||
/// travel the SDK channel (a condition's method audits), on the SDK's per-node delivery task, where the
|
||||
/// inherited role-permission check already takes <c>Lock</c> (via <c>GetNodeMetadata</c>); the scope
|
||||
/// resolution here takes it the same way.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
private bool IsNotifierEventVisibleTo(IEventMonitoredItem item, IFilterTarget e)
|
||||
@@ -6146,13 +6354,15 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3
|
||||
/// <returns>Good to deliver; otherwise the veto.</returns>
|
||||
/// <remarks>
|
||||
/// <para>
|
||||
/// The SDK's <c>MonitoredNode2</c> asks this, per event and per item, before it queues an event to an item
|
||||
/// attached to one of this manager's notifiers. The Server-object path (#586) and ConditionRefresh decide
|
||||
/// the same question themselves, so a folder's subscribers stop receiving a moved alarm (or a native alarm
|
||||
/// whose reference was dropped) the moment the generation says so, not when the rebuild lands.
|
||||
/// Condition events to an item attached to one of this manager's notifiers are decided here per event and per
|
||||
/// item by <see cref="DeliverToNotifierItems"/> (#855, uncached); the SDK's <c>MonitoredNode2</c> still asks it
|
||||
/// for what travels its channel (a condition's method audit events). The Server-object path (#586) and
|
||||
/// ConditionRefresh decide the same question themselves, so a folder's subscribers stop receiving a moved
|
||||
/// alarm (or a native alarm whose reference was dropped) the moment the generation says so, not when the
|
||||
/// rebuild lands.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// ⚠️ The SDK MEMOIZES the verdict per <c>(item, event type, SourceNode)</c> until the monitored node's
|
||||
/// ⚠️ On the SDK channel path the SDK MEMOIZES the verdict per <c>(item, event type, SourceNode)</c> until the monitored node's
|
||||
/// permission generation moves, so <see cref="InvalidateFolderEventVerdicts"/> must run whenever the inputs
|
||||
/// change: a new authority generation or authorizer, the revalidation tick, and the end of every apply. A
|
||||
/// condition's SourceNode is the condition itself, so a verdict is never shared between two conditions.
|
||||
@@ -6278,9 +6488,9 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3
|
||||
// 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.
|
||||
// Items on one of this manager's own notifiers go through RefreshDirectItems in both modes since #855 — the
|
||||
// base's loop (a stood-down item is off its notifier's fan-out, so not refreshed) plus the #826 decision with
|
||||
// enforcement on, and the same cut against deferred events as the Server object.
|
||||
/// <inheritdoc />
|
||||
public override ServiceResult ConditionRefresh(OperationContext context, IList<IEventMonitoredItem> monitoredItems)
|
||||
{
|
||||
@@ -6301,11 +6511,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3
|
||||
|
||||
var enforced = AclEnforced;
|
||||
var systemContext = SystemContext.Copy(context);
|
||||
if (direct.Count > 0)
|
||||
{
|
||||
if (enforced) RefreshDirectItems(systemContext, direct);
|
||||
else base.ConditionRefresh(context, direct);
|
||||
}
|
||||
if (direct.Count > 0) RefreshDirectItems(systemContext, direct, enforced);
|
||||
|
||||
if (serverItems.Count > 0) RefreshServerObjectItems(systemContext, serverItems, enforced);
|
||||
return ServiceResult.Good;
|
||||
@@ -6339,7 +6545,9 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3
|
||||
/// 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.
|
||||
/// refresh had just dropped re-added, until the newest drained. (Since #855 folder items are no longer a
|
||||
/// channel, so nothing of ours defers today; the cut stays as the safety net, and folder items share it —
|
||||
/// <see cref="RefreshDirectItems"/>.)
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// <b>Common case — nothing deferred: unchanged, and off <c>Lock</c>.</b> The replay is the notifier walk the SDK
|
||||
@@ -6353,10 +6561,10 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3
|
||||
/// <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 /
|
||||
/// with its <see cref="RefreshCut"/> installed under the item's <see cref="EventItemState.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
|
||||
/// <see cref="EventItemState.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"/>)
|
||||
@@ -6406,17 +6614,12 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3
|
||||
{
|
||||
// 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();
|
||||
}
|
||||
}
|
||||
StopPendingReplay(item);
|
||||
|
||||
foreach (var (e, _, scopes) in replay)
|
||||
{
|
||||
// #855 review 3: never replay a condition a deploy removed or replaced since the walk collected it.
|
||||
if (ConditionOf(e) is AlarmConditionState walked && !IsLiveCondition(walked)) continue;
|
||||
if (scopes is not null && !IsConditionVisibleTo(item, scopes)) continue;
|
||||
if (ServiceResult.IsBad(ValidateEventRolePermissions(item, e))) continue;
|
||||
item.QueueEvent(e);
|
||||
@@ -6472,7 +6675,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3
|
||||
}
|
||||
}
|
||||
|
||||
var states = items.Select(AllEventsItemStateOf).ToList();
|
||||
var states = items.Select(EventItemStateOf).ToList();
|
||||
var perItem = new List<(IFilterTarget Event, AlarmConditionState Condition)>[items.Count];
|
||||
var generation = Interlocked.Increment(ref _refreshGeneration);
|
||||
// #854 re-review: a throw between installing the "replay pending" markers and finishing the replay must not leave
|
||||
@@ -6540,20 +6743,69 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3
|
||||
}
|
||||
}
|
||||
|
||||
// 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));
|
||||
// #855 review: the verdicts above were taken before Lock; a deploy may have revoked a grant since. Each replay entry
|
||||
// is re-decided right before it is queued (scopes resolved once per condition).
|
||||
var scopeCache = new Dictionary<NodeId, List<NodeId>>();
|
||||
QueueFrozenReplay(
|
||||
items, states, perItem.Select(entries => entries.Select(x => (x.Event, x.Condition.NodeId)).ToList()).ToArray(),
|
||||
generation,
|
||||
(item, e) =>
|
||||
{
|
||||
// #855 review 2: never replay a condition a deploy has removed or replaced since the cut — its frozen
|
||||
// state would be a ghost (and, with the ACL on, a replaced scripted alarm would be judged on its old folder).
|
||||
if (ConditionOf(e) is { } frozen && !IsLiveCondition(frozen)) return false;
|
||||
if (enforced)
|
||||
{
|
||||
if (ConditionOf(e) is not { } condition) return false;
|
||||
if (!scopeCache.TryGetValue(condition.NodeId, out var scopes))
|
||||
scopeCache[condition.NodeId] = scopes = ConditionVisibilityScopes(condition);
|
||||
if (!IsConditionVisibleTo(item, scopes)) return false;
|
||||
}
|
||||
|
||||
return !ServiceResult.IsBad(ValidateEventRolePermissions(item, e));
|
||||
});
|
||||
}
|
||||
finally
|
||||
{
|
||||
ClearPendingReplays(states, generation);
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Queues a cut-path refresh's frozen replay to its items OFF <c>Lock</c>, one replayed event per hold of the item
|
||||
/// gate (#854; folder items too since #855): 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.
|
||||
/// </summary>
|
||||
/// <param name="allowed">
|
||||
/// The item's verdict for one entry, asked right before the entry's gate hold (never under it: it takes <c>Lock</c>),
|
||||
/// against the CURRENT generation — the verdicts the cut was built on were taken earlier (#855 review). An entry
|
||||
/// denied here is dropped AND removed from the cut's re-sent set, so the condition counts as "omitted for the
|
||||
/// grants" and its later events are left to the ACL rather than withheld as superseded.
|
||||
/// </param>
|
||||
private static void QueueFrozenReplay(
|
||||
List<IEventMonitoredItem> items, List<EventItemState> states, List<(IFilterTarget Event, NodeId ConditionId)>[] perItem,
|
||||
int generation, Func<IEventMonitoredItem, IFilterTarget, bool> allowed)
|
||||
{
|
||||
for (var i = 0; i < items.Count; i++)
|
||||
{
|
||||
var item = items[i];
|
||||
var state = states[i];
|
||||
foreach (var (e, condition) in perItem[i])
|
||||
foreach (var (e, conditionId) in perItem[i])
|
||||
{
|
||||
if (state.PendingReplay != generation) break;
|
||||
var permitted = allowed(item, e);
|
||||
lock (state.Gate)
|
||||
{
|
||||
if (state.PendingReplay != generation) break;
|
||||
if (state.NewerSinceCut.Contains(condition.NodeId)) continue;
|
||||
if (!permitted)
|
||||
{
|
||||
state.Cut?.Refreshed.Remove(conditionId);
|
||||
continue;
|
||||
}
|
||||
|
||||
if (state.NewerSinceCut.Contains(conditionId)) continue;
|
||||
item.QueueEvent(e);
|
||||
}
|
||||
}
|
||||
@@ -6567,51 +6819,234 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
finally
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// #854 re-review: a throw between installing the "replay pending" markers and finishing the replay must not leave a
|
||||
/// marker set (every sync event to that item would keep taking the gate until its next refresh).
|
||||
/// </summary>
|
||||
private static void ClearPendingReplays(List<EventItemState> states, int generation)
|
||||
{
|
||||
foreach (var state in states)
|
||||
{
|
||||
foreach (var state in states)
|
||||
lock (state.Gate)
|
||||
{
|
||||
lock (state.Gate)
|
||||
{
|
||||
if (state.PendingReplay != generation) continue;
|
||||
state.PendingReplay = 0;
|
||||
state.NewerSinceCut.Clear();
|
||||
}
|
||||
if (state.PendingReplay != generation) continue;
|
||||
state.PendingReplay = 0;
|
||||
state.NewerSinceCut.Clear();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// ConditionRefresh for event items on this manager's own notifiers, under enforcement (Gitea #826): the SDK
|
||||
/// base's loop exactly — an item no longer attached (stood down by revalidation, or tombstoned) is not
|
||||
/// refreshed — plus the per-event decision live delivery makes (<see cref="IsNotifierEventVisibleTo"/>), so a
|
||||
/// refresh cannot replay to a folder's subscriber an alarm that folder's live events withhold.
|
||||
/// Stops a cut-path replay of <paramref name="item"/> still being queued by a concurrent refresh: a live replay that
|
||||
/// is about to be queued is newer, and the frozen one must not follow it.
|
||||
/// </summary>
|
||||
private void StopPendingReplay(IEventMonitoredItem item)
|
||||
{
|
||||
if (!_eventItemStates.TryGetValue(item, out var pending)) return;
|
||||
lock (pending.Gate)
|
||||
{
|
||||
pending.PendingReplay = 0;
|
||||
pending.NewerSinceCut.Clear();
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// ConditionRefresh (and ConditionRefresh2) for event items on this manager's own notifiers — equipment and raw device
|
||||
/// folders — flag on and off: never overtaken by a deferred event older than what it replays (Gitea #855), the folder
|
||||
/// twin of <see cref="RefreshServerObjectItems"/>.
|
||||
/// </summary>
|
||||
/// <param name="systemContext">The refresh's system context.</param>
|
||||
/// <param name="items">The direct (non-Server-object) items being refreshed.</param>
|
||||
private void RefreshDirectItems(ServerSystemContext systemContext, List<IEventMonitoredItem> items)
|
||||
/// <param name="enforced">Whether the data-plane ACL is enforced.</param>
|
||||
/// <remarks>
|
||||
/// <para>
|
||||
/// <b>Nothing deferred (the only case today — no notifier of ours defers since #855): unchanged, off
|
||||
/// <c>Lock</c>.</b> The SDK base's loop — an item no longer attached (stood down by revalidation, or tombstoned)
|
||||
/// is not refreshed; the item's notifier is walked and each live condition queued through the SDK's role check —
|
||||
/// plus, under enforcement, the per-event decision live delivery makes (<see cref="IsNotifierEventVisibleTo"/>,
|
||||
/// #826), so a refresh cannot replay to a folder's subscriber an alarm that folder's live events withhold. It
|
||||
/// stands only if no deferral episode existed at any moment of it; the closing check takes <c>Lock</c> once, so a
|
||||
/// report whose walk suspended (it marks the event before suspending, under <c>Lock</c>) has been counted by then.
|
||||
/// Otherwise the cut path runs as well, and its replay — later inside the same RefreshStart..RefreshEnd —
|
||||
/// supersedes anything that slipped in between.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// <b>Deferring — the cut</b>, exactly as on the Server object: verdicts per (item, condition) first, off
|
||||
/// <c>Lock</c>, from a live walk; then one <c>Lock</c> hold takes the watermark, the episode, the Retain set and,
|
||||
/// per item, the same walk as frozen snapshots, and installs the cut and the "replay pending" marker under the
|
||||
/// item's gate; the replay is queued off <c>Lock</c> (<see cref="QueueFrozenReplay"/>). Deferred events reaching
|
||||
/// the item afterwards are judged against the cut (<see cref="QueueConditionEventToItem"/>).
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
private void RefreshDirectItems(ServerSystemContext systemContext, List<IEventMonitoredItem> items, bool enforced)
|
||||
{
|
||||
foreach (var item in items)
|
||||
long episodeBefore;
|
||||
bool deferringBefore;
|
||||
lock (_eventReportGate)
|
||||
{
|
||||
NodeState? notifier;
|
||||
lock (Lock)
|
||||
episodeBefore = _deferralEpisode;
|
||||
deferringBefore = _deferredEventReports > 0;
|
||||
}
|
||||
|
||||
if (!deferringBefore)
|
||||
{
|
||||
foreach (var item in items)
|
||||
{
|
||||
if (!MonitoredItems.ContainsKey(item.Id)) continue;
|
||||
notifier = (item.ManagerHandle as NodeHandle)?.Node;
|
||||
var notifier = AttachedNotifierOf(item);
|
||||
if (notifier is null) continue;
|
||||
|
||||
StopPendingReplay(item);
|
||||
var events = new List<IFilterTarget>();
|
||||
notifier.ConditionRefresh(systemContext, events, true);
|
||||
DirectRefreshProbe?.Invoke(false, Monitor.IsEntered(Lock));
|
||||
foreach (var e in events)
|
||||
{
|
||||
// #855 review 3: never replay a condition a deploy removed or replaced since the walk collected it.
|
||||
if (ConditionOf(e) is AlarmConditionState walked && !IsLiveCondition(walked)) continue;
|
||||
if (!IsDirectRefreshEventAllowed(item, e, enforced)) continue;
|
||||
item.QueueEvent(e);
|
||||
}
|
||||
}
|
||||
|
||||
if (notifier is null) continue;
|
||||
// Lock first: a report that suspended under it has registered its deferral by the time Lock is free.
|
||||
lock (Lock)
|
||||
{
|
||||
lock (_eventReportGate)
|
||||
{
|
||||
if (_deferralEpisode == episodeBefore && _deferredEventReports == 0) return;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
RefreshDirectItemsWithCut(systemContext, items, enforced);
|
||||
}
|
||||
|
||||
/// <summary>The notifier an attached direct item subscribes to, or null for one that is not attached (not refreshed).</summary>
|
||||
private NodeState? AttachedNotifierOf(IEventMonitoredItem item)
|
||||
{
|
||||
lock (Lock)
|
||||
{
|
||||
return MonitoredItems.ContainsKey(item.Id) ? (item.ManagerHandle as NodeHandle)?.Node : null;
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>A direct item's refresh verdict for one replayed event: the #826 decision under enforcement, then the SDK's.</summary>
|
||||
private bool IsDirectRefreshEventAllowed(IEventMonitoredItem item, IFilterTarget e, bool enforced) =>
|
||||
(!enforced || IsNotifierEventVisibleTo(item, e)) && !ServiceResult.IsBad(ValidateEventRolePermissions(item, e));
|
||||
|
||||
/// <summary>
|
||||
/// Test probe (#855): called by a direct-item 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>? DirectRefreshProbe { get; set; }
|
||||
|
||||
/// <summary>The cut path of <see cref="RefreshDirectItems"/> — see there.</summary>
|
||||
private void RefreshDirectItemsWithCut(ServerSystemContext systemContext, List<IEventMonitoredItem> items, bool enforced)
|
||||
{
|
||||
// Verdicts per (item, condition) before Lock, from a live walk: they depend on the event type, source, notifier
|
||||
// wiring and grants, not on the condition's state.
|
||||
var allowed = new HashSet<(IEventMonitoredItem, NodeState)>();
|
||||
// Per (item, condition): a condition one item's walk saw is still undecided for another item (#855 review 4).
|
||||
var known = new HashSet<(IEventMonitoredItem, NodeState)>();
|
||||
foreach (var item in items)
|
||||
{
|
||||
if (AttachedNotifierOf(item) is not { } notifier) continue;
|
||||
var events = new List<IFilterTarget>();
|
||||
notifier.ConditionRefresh(systemContext, events, true);
|
||||
foreach (var e in events)
|
||||
{
|
||||
if (!IsNotifierEventVisibleTo(item, e)) continue;
|
||||
if (ServiceResult.IsBad(ValidateEventRolePermissions(item, e))) continue;
|
||||
item.QueueEvent(e);
|
||||
if (ConditionOf(e) is not { } node) continue;
|
||||
known.Add((item, node));
|
||||
if (IsDirectRefreshEventAllowed(item, e, enforced)) allowed.Add((item, node));
|
||||
}
|
||||
}
|
||||
|
||||
var states = items.Select(EventItemStateOf).ToList();
|
||||
var perItem = new List<(IFilterTarget Event, NodeId ConditionId)>[items.Count];
|
||||
var generation = Interlocked.Increment(ref _refreshGeneration);
|
||||
try
|
||||
{
|
||||
lock (Lock)
|
||||
{
|
||||
var watermark = Interlocked.Read(ref _conditionEventSequence);
|
||||
long episode;
|
||||
bool deferring;
|
||||
lock (_eventReportGate)
|
||||
{
|
||||
episode = _deferralEpisode;
|
||||
deferring = _deferredEventReports > 0;
|
||||
}
|
||||
|
||||
var retained = new HashSet<NodeId>();
|
||||
foreach (var condition in _alarmConditions.Values)
|
||||
{
|
||||
if (condition.Retain?.Value == true) retained.Add(condition.NodeId);
|
||||
}
|
||||
|
||||
for (var i = 0; i < items.Count; i++)
|
||||
{
|
||||
var item = items[i];
|
||||
var entries = new List<(IFilterTarget Event, NodeId ConditionId)>();
|
||||
perItem[i] = entries;
|
||||
var state = states[i];
|
||||
|
||||
// Lock is held, so this is the attachment the walk below sees. An unattached item is not refreshed and
|
||||
// gets no cut (it receives nothing until re-attached, and then every event is decided as it arrives).
|
||||
var notifier = MonitoredItems.ContainsKey(item.Id) ? (item.ManagerHandle as NodeHandle)?.Node : null;
|
||||
if (notifier is null)
|
||||
{
|
||||
lock (state.Gate)
|
||||
{
|
||||
state.PendingReplay = 0;
|
||||
state.NewerSinceCut.Clear();
|
||||
}
|
||||
|
||||
continue;
|
||||
}
|
||||
|
||||
// Frozen at the watermark: the replay is exactly the state every event at or below it describes.
|
||||
var live = new List<IFilterTarget>();
|
||||
notifier.ConditionRefresh(SystemContext, live, true);
|
||||
var refreshed = new HashSet<NodeId>();
|
||||
foreach (var target in live)
|
||||
{
|
||||
if (target is not BaseInstanceState instance) continue;
|
||||
// A condition materialised since the verdicts were taken is decided here (rare).
|
||||
if (!allowed.Contains((item, instance)))
|
||||
{
|
||||
if (known.Contains((item, instance))) continue;
|
||||
if (!IsDirectRefreshEventAllowed(item, target, enforced)) continue;
|
||||
}
|
||||
|
||||
var snapshot = new InstanceStateSnapshot();
|
||||
snapshot.Initialize(SystemContext, instance);
|
||||
entries.Add((snapshot, instance.NodeId));
|
||||
refreshed.Add(instance.NodeId);
|
||||
}
|
||||
|
||||
lock (state.Gate)
|
||||
{
|
||||
state.Cut = deferring ? new RefreshCut(watermark, episode, refreshed, retained) : null;
|
||||
state.PendingWatermark = watermark;
|
||||
state.NewerSinceCut.Clear();
|
||||
state.PendingReplay = generation;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
DirectRefreshProbe?.Invoke(true, Monitor.IsEntered(Lock));
|
||||
// #855 review: re-decided per entry right before queueing — a deploy may have revoked since the verdicts.
|
||||
QueueFrozenReplay(
|
||||
items, states, perItem, generation,
|
||||
(item, e) => (ConditionOf(e) is not AlarmConditionState frozen || IsLiveCondition(frozen))
|
||||
&& IsDirectRefreshEventAllowed(item, e, enforced));
|
||||
}
|
||||
finally
|
||||
{
|
||||
ClearPendingReplays(states, generation);
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>The condition a refresh event describes, or null when it names none (withheld: fail closed).</summary>
|
||||
|
||||
+15
-55
@@ -24,16 +24,18 @@ namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests;
|
||||
/// 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.
|
||||
/// The overload is produced deterministically: delivery is stalled at the EqA folder (<see cref="NotifierStall"/>), so the
|
||||
/// condition's walk suspends there and its later events are deferred — without holding the node manager's <c>Lock</c>,
|
||||
/// which the refresh needs. (Before Gitea #855 this held the folder item's SDK lock so the folder's 4096-slot channel
|
||||
/// filled; folder items are now queued per item by the reporter, and no notifier of ours defers on its own.) The
|
||||
/// folder-item twin of these tests is <see cref="FolderConditionRefreshWireTests"/>.
|
||||
/// </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;
|
||||
// Every event after the stalled one is deferred; < 1024 pending, so nothing coalesces.
|
||||
private const int FloodSize = 500;
|
||||
|
||||
/// <param name="enforced">Per-folder / Server-object ACL on or off (off is the production default).</param>
|
||||
/// <param name="finalRetained">
|
||||
@@ -84,7 +86,7 @@ public sealed class ConditionRefreshDeferralWireTests
|
||||
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)))
|
||||
using (var stall = NotifierStall.Hold(rig.Nm, EqA))
|
||||
{
|
||||
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");
|
||||
@@ -200,7 +202,7 @@ public sealed class ConditionRefreshDeferralWireTests
|
||||
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)))
|
||||
using (var stall = NotifierStall.Hold(rig.Nm, EqA))
|
||||
{
|
||||
for (var i = 0; i < FloodSize; i++) rig.Project(native: true, Active($"f{i}"));
|
||||
rig.Project(native: true, Active("final"));
|
||||
@@ -271,7 +273,7 @@ public sealed class ConditionRefreshDeferralWireTests
|
||||
rig.Sink.WriteAlarmCondition(Free, Active("x2"), DateTime.UtcNow, AddressSpaceRealm.Raw);
|
||||
};
|
||||
|
||||
using (var stall = ItemLockStall.Hold(FindServerItem(rig.Nm, onA.Status.Id)))
|
||||
using (var stall = NotifierStall.Hold(rig.Nm, EqA))
|
||||
{
|
||||
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");
|
||||
@@ -345,7 +347,7 @@ public sealed class ConditionRefreshDeferralWireTests
|
||||
proceed.Wait(TimeSpan.FromSeconds(30));
|
||||
};
|
||||
|
||||
using (ItemLockStall.Hold(FindServerItem(rig.Nm, onA.Status.Id)))
|
||||
using (NotifierStall.Hold(rig.Nm, EqA))
|
||||
{
|
||||
for (var i = 0; i < FloodSize; i++) rig.Project(native: true, Active($"f{i}"));
|
||||
rig.Project(native: true, Active("final"));
|
||||
@@ -416,7 +418,6 @@ public sealed class ConditionRefreshDeferralWireTests
|
||||
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;
|
||||
@@ -440,7 +441,7 @@ public sealed class ConditionRefreshDeferralWireTests
|
||||
{
|
||||
while (!stop.IsCancellationRequested)
|
||||
{
|
||||
using (ItemLockStall.Hold(item))
|
||||
using (NotifierStall.Hold(rig.Nm, EqA))
|
||||
{
|
||||
Thread.Sleep(400);
|
||||
if (rig.Nm.DeferredEventReports > 0) Interlocked.Increment(ref deferredSeen);
|
||||
@@ -568,7 +569,7 @@ public sealed class ConditionRefreshDeferralWireTests
|
||||
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;
|
||||
var stall = deferring ? NotifierStall.Hold(rig.Nm, EqA) : null;
|
||||
try
|
||||
{
|
||||
if (deferring)
|
||||
@@ -622,7 +623,7 @@ public sealed class ConditionRefreshDeferralWireTests
|
||||
}
|
||||
|
||||
/// <summary>The server-side Server-object (all-events) item behind a client item.</summary>
|
||||
private static IEventMonitoredItem FindAllEventsItem(OtOpcUaNodeManager nm, uint id)
|
||||
internal static IEventMonitoredItem FindAllEventsItem(OtOpcUaNodeManager nm, uint id)
|
||||
{
|
||||
var map = (IDictionary)typeof(OtOpcUaNodeManager)
|
||||
.GetField("_allEventsItems", BindingFlags.Instance | BindingFlags.NonPublic)!
|
||||
@@ -636,7 +637,7 @@ public sealed class ConditionRefreshDeferralWireTests
|
||||
}
|
||||
|
||||
/// <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)
|
||||
internal static IEventMonitoredItem FindServerItem(OtOpcUaNodeManager nm, uint id)
|
||||
{
|
||||
var map = (IDictionary)typeof(OtOpcUaNodeManager)
|
||||
.GetField("_eventItems", BindingFlags.Instance | BindingFlags.NonPublic)!
|
||||
@@ -664,45 +665,4 @@ public sealed class ConditionRefreshDeferralWireTests
|
||||
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();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
+18
-13
@@ -27,13 +27,16 @@ namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests;
|
||||
/// <item>no ACL-denied event reaches an item during the overflow, including deferred events whose grant was
|
||||
/// revoked while they waited.</item>
|
||||
/// </list>
|
||||
/// Since Gitea #855 the folder items are served per item by the reporter (as the Server-object items always were),
|
||||
/// so the flood no longer defers at all — the first test pins that — and the revocation test stalls the folder's
|
||||
/// delivery (<see cref="NotifierStall"/>) to get a deferred backlog.
|
||||
/// </summary>
|
||||
public sealed class EventQueueOverflowAclWireTests
|
||||
{
|
||||
private const string EqA = ServerEventRig.EqA;
|
||||
private const string EqB = ServerEventRig.EqB;
|
||||
|
||||
// Native events per flood: past the SDK's 4096-slot per-notifier channel, so the native condition defers.
|
||||
// Native events per flood: past the SDK's 4096-slot per-notifier channel (which deferred them before #855).
|
||||
private const int PerCondition = 6000;
|
||||
|
||||
/// <param name="singleHold">
|
||||
@@ -150,13 +153,12 @@ public sealed class EventQueueOverflowAclWireTests
|
||||
readLatencies.Max().ShouldBeLessThan(TimeSpan.FromSeconds(5), "a Read must not queue behind the whole flood");
|
||||
(await TimedReadAsync(anonymous, eqA, ct)).ShouldBeLessThan(TimeSpan.FromSeconds(2), "the node manager is free afterwards");
|
||||
|
||||
// Memory: the node manager's own overflow store stayed within its per-condition budget (queued + in flight)…
|
||||
Volatile.Read(ref peakDeferred).ShouldBeLessThanOrEqualTo(
|
||||
2 * (OtOpcUaNodeManager.MaxDeferredEventsPerCondition + 1), "deferral is bounded per condition");
|
||||
if (singleHold) rig.Nm.DiscardedEventReports.ShouldBeGreaterThan(0, "the native tail overflowed and coalesced");
|
||||
// Memory: the node manager held nothing of its own for the overflow (#855: no notifier of ours defers, so the
|
||||
// item queues are the only store, bounded below)…
|
||||
Volatile.Read(ref peakDeferred).ShouldBe(0, "folder items are served per item by the reporter: nothing defers");
|
||||
rig.Nm.DiscardedEventReports.ShouldBe(0);
|
||||
|
||||
// …let the SDK channel finish draining into the (still unpublished) item queues, then publish.
|
||||
await Task.Delay(TimeSpan.FromSeconds(2), ct);
|
||||
// …then publish the (still unpublished) item queues.
|
||||
await keySub.SetPublishingModeAsync(true, ct);
|
||||
var lastNative = $"n{PerCondition - 1}";
|
||||
await RecordedEvents.WaitUntilAsync(() => keyEvents.Has(onB, lastNative) && keyEvents.Has(keyServer, lastNative));
|
||||
@@ -165,10 +167,9 @@ public sealed class EventQueueOverflowAclWireTests
|
||||
foreach (var item in new[] { onB, keyServer })
|
||||
{
|
||||
var received = keyEvents.Of(item);
|
||||
// ACL: not one event of the denied (EqA-owned) scripted alarm, overflow or not. (Under a single hold the
|
||||
// deferred native tail drains last and evicts every synchronously-queued event from the 1000-slot queue, so
|
||||
// there this check cannot see a leak; the "after-s" leg below is the falsifier for that mode — verified by
|
||||
// running with enforcement off: chunked fails here, single-hold fails at "after-s".)
|
||||
// ACL: not one event of the denied (EqA-owned) scripted alarm, overflow or not. (The 1000-slot queue keeps
|
||||
// only the last events of the flood, which interleave n/s, so a leak shows here in both modes; the
|
||||
// "after-s" leg below is a second falsifier.)
|
||||
received.Where(f => RecordedEvents.Message(f)?.StartsWith('s') == true)
|
||||
.ShouldBeEmpty($"{item.DisplayName}: an ACL-denied event leaked during the overflow");
|
||||
|
||||
@@ -211,7 +212,7 @@ public sealed class EventQueueOverflowAclWireTests
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// A grant revoked while a deferred backlog waits behind a full notifier channel: the deferred events are decided
|
||||
/// A grant revoked while a deferred backlog waits behind a stalled notifier: the deferred events are decided
|
||||
/// at delivery against the CURRENT generation, so none of them — nor any event reported after the revocation —
|
||||
/// reaches the revoked session's folder or Server-object item, while a still-granted session sharing the same
|
||||
/// folder channel receives the whole tail (positive control).
|
||||
@@ -249,6 +250,9 @@ public sealed class EventQueueOverflowAclWireTests
|
||||
await anonSub.ApplyChangesAsync(ct);
|
||||
anonA.Status.Error.ShouldBeNull();
|
||||
|
||||
// The EqA folder's delivery stalls (#855: nothing of ours defers on its own any more), so the native condition's
|
||||
// walk suspends there and the rest of the flood is deferred behind it.
|
||||
using var stall = NotifierStall.Hold(rig.Nm, EqA);
|
||||
var flood = Task.Factory.StartNew(() =>
|
||||
{
|
||||
lock (rig.Nm.Lock)
|
||||
@@ -276,11 +280,12 @@ public sealed class EventQueueOverflowAclWireTests
|
||||
throw new ShouldAssertException("revoking a grant under Lock while events were deferred stalled (deadlock)");
|
||||
}
|
||||
|
||||
stall.Release();
|
||||
await RecordedEvents.WaitUntilAsync(() => rig.Nm.DeferredEventReports == 0);
|
||||
await Task.Delay(TimeSpan.FromSeconds(2), ct);
|
||||
await keySub.SetPublishingModeAsync(true, ct);
|
||||
|
||||
// Positive control: the still-granted session on the SAME folder channel gets the whole deferred tail.
|
||||
// Positive control: the still-granted session on the SAME folder gets the whole deferred tail.
|
||||
var lastNative = $"n{PerCondition - 1}";
|
||||
await RecordedEvents.WaitUntilAsync(() =>
|
||||
anonEvents.Has(anonA, lastNative) && anonEvents.Has(anonA, "after-revoke")
|
||||
|
||||
+841
@@ -0,0 +1,841 @@
|
||||
using System.Diagnostics;
|
||||
using System.Globalization;
|
||||
using Opc.Ua;
|
||||
using Opc.Ua.Client;
|
||||
using Shouldly;
|
||||
using Xunit;
|
||||
using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
|
||||
using ZB.MOM.WW.OtOpcUa.Configuration.Entities;
|
||||
using ZB.MOM.WW.OtOpcUa.Configuration.Enums;
|
||||
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
||||
using ZB.MOM.WW.OtOpcUa.Core.Authorization;
|
||||
using MonitoredItem = Opc.Ua.Client.MonitoredItem;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests;
|
||||
|
||||
/// <summary>
|
||||
/// Gitea #855: a <c>ConditionRefresh</c> of an event item on an equipment / raw device FOLDER gets the guarantee the
|
||||
/// Server object got in #854 — no condition event older than what the refresh replayed reaches the item after
|
||||
/// RefreshEnd, per-condition order is kept, the per-folder ACL (#826) decides every event and every replayed condition,
|
||||
/// and neither a reporter nor the refresh ever waits on a stalled delivery.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <para>
|
||||
/// Folder items are now served per item by the node manager (no SDK channel), so in production no notifier defers
|
||||
/// at all; <see cref="Production_overload_never_defers_and_never_overtakes_a_folder_refresh"/> pins that under a
|
||||
/// flood with concurrent refreshes. Every other test produces the deferral the cut exists for by stalling delivery
|
||||
/// at a notifier (<see cref="NotifierStall"/>), the way a full SDK channel used to.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// The rig's native condition is wired into EqA and EqB and hangs under the raw device folder, so its walk is
|
||||
/// condition → EqA → … → device folder; a stall at EqA suspends it before EqA's items and every later hop.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
public sealed class FolderConditionRefreshWireTests
|
||||
{
|
||||
private const string EqA = ServerEventRig.EqA;
|
||||
|
||||
// Every event after the stalled one is deferred; < 1024 pending, so nothing coalesces.
|
||||
private const int FloodSize = 500;
|
||||
|
||||
/// <param name="enforced">Per-folder ACL on or off (off is the production default).</param>
|
||||
/// <param name="finalRetained">Whether the condition is still Retain at refresh time (re-sent) or cleared (omitted).</param>
|
||||
/// <param name="refresh2">ConditionRefresh2 on the folder item instead of ConditionRefresh on its subscription.</param>
|
||||
/// <param name="modify">
|
||||
/// The folder item is modified (ModifyMonitoredItems re-subscribes it to its notifier) between the refresh and the
|
||||
/// drain: its cut must survive, as the Server object's does across an unsubscribe + resubscribe.
|
||||
/// </param>
|
||||
[Theory]
|
||||
[InlineData(false, true, false, false)]
|
||||
[InlineData(true, true, false, false)]
|
||||
[InlineData(false, false, false, false)]
|
||||
[InlineData(true, false, false, false)]
|
||||
[InlineData(false, true, true, false)]
|
||||
[InlineData(true, false, true, false)]
|
||||
[InlineData(false, true, false, true)]
|
||||
[InlineData(true, false, false, true)]
|
||||
public async Task Deferred_events_older_than_a_refresh_never_reach_the_refreshed_folder_item_after_RefreshEnd(
|
||||
bool enforced, bool finalRetained, bool refresh2, bool modify)
|
||||
{
|
||||
var ct = TestContext.Current.CancellationToken;
|
||||
await using var rig = await ServerEventRig.StartAsync("Refresh855", encrypted: false, ct);
|
||||
rig.KeyRoles = [];
|
||||
rig.MaterialiseNative();
|
||||
rig.Enforce(enforced, [ServerEventRig.Key()], KeyEquipmentGrant(EqA));
|
||||
|
||||
// A second, non-refreshing subscriber on the SAME folder: it must keep the whole deferred tail.
|
||||
using var otherSession = await rig.ConnectAsync(anonymous: false, ct);
|
||||
var otherEvents = new RecordedEvents();
|
||||
var otherSubscription = await RecordedEvents.SubscribeAsync(otherSession, otherEvents, ct);
|
||||
var otherOnA = RecordedEvents.AddEventItem(otherSubscription, ServerEventRig.UnsNodeId(otherSession, EqA));
|
||||
await otherSubscription.ApplyChangesAsync(ct);
|
||||
otherOnA.Status.Error.ShouldBeNull();
|
||||
|
||||
using var session = await rig.ConnectAsync(anonymous: false, ct);
|
||||
var events = new RecordedEvents();
|
||||
var subscription = await RecordedEvents.SubscribeAsync(session, events, ct);
|
||||
var onA = RecordedEvents.AddEventItem(subscription, ServerEventRig.UnsNodeId(session, EqA));
|
||||
await subscription.ApplyChangesAsync(ct);
|
||||
onA.Status.Error.ShouldBeNull();
|
||||
|
||||
rig.Project(native: true, Active("seed"));
|
||||
await RecordedEvents.WaitUntilAsync(() => events.Has(onA, "seed") && otherEvents.Has(otherOnA, "seed"));
|
||||
|
||||
using (var stall = NotifierStall.Hold(rig.Nm, EqA))
|
||||
{
|
||||
for (var i = 0; i < FloodSize; i++) rig.Project(native: true, Active($"f{i}"));
|
||||
rig.Nm.DeferredEventReports.ShouldBeGreaterThan(0, "the stall must actually have deferred events");
|
||||
rig.Project(native: true, finalRetained ? Active("final") : Cleared("final"));
|
||||
|
||||
var refresh = Stopwatch.StartNew();
|
||||
await RefreshAsync(session, subscription, onA, refresh2, ct);
|
||||
refresh.Elapsed.ShouldBeLessThan(TimeSpan.FromSeconds(10), "ConditionRefresh must not wait for the stalled delivery");
|
||||
await RecordedEvents.WaitUntilAsync(() => events.HasType(onA, ObjectTypeIds.RefreshEndEventType));
|
||||
rig.Nm.DeferredEventReports.ShouldBeGreaterThan(0, "the deferred events were still waiting at RefreshEnd");
|
||||
rig.Nm.ActiveRefreshCuts.ShouldBe(1, "only the refreshed folder item holds a cut");
|
||||
|
||||
if (modify)
|
||||
{
|
||||
onA.QueueSize = 1001;
|
||||
await subscription.ApplyChangesAsync(ct);
|
||||
onA.Status.Error.ShouldBeNull();
|
||||
}
|
||||
|
||||
stall.Release();
|
||||
}
|
||||
|
||||
await RecordedEvents.WaitUntilAsync(() => rig.Nm.DeferredEventReports == 0);
|
||||
await RecordedEvents.WaitUntilAsync(() => rig.Nm.ActiveRefreshCuts == 0);
|
||||
rig.Project(native: true, Active("after"));
|
||||
await RecordedEvents.WaitUntilAsync(() => events.Has(onA, "after") && otherEvents.Has(otherOnA, "after"));
|
||||
await Task.Delay(TimeSpan.FromMilliseconds(300), ct);
|
||||
|
||||
var (replayed, afterEnd) = Window(events.Of(onA));
|
||||
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");
|
||||
afterEnd.ShouldBe(["final", "after"], "stale Retain=true events would re-add the alarm; the Retain=false one agrees");
|
||||
events.Of(onA).Last(f => RecordedEvents.Message(f) == "final")[RecordedEvents.RetainField].GetBoolean().ShouldBeFalse();
|
||||
}
|
||||
|
||||
rig.Nm.RefreshSupersededDeliveries.ShouldBeGreaterThan(0);
|
||||
|
||||
// The other subscriber on the same folder did not refresh: it keeps every deferred event, in order.
|
||||
var flood = Indices(otherEvents.Of(otherOnA), 'f');
|
||||
flood.ShouldBe(Enumerable.Range(0, FloodSize).ToList(), "the non-refreshing folder subscriber keeps the whole tail, in order");
|
||||
otherEvents.Has(otherOnA, "final").ShouldBeTrue();
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// The race the item gate closes, for a folder item: a deferred event is between its supersession check and its
|
||||
/// queueing to the folder item when the refresh arrives. The refresh must wait for that one delivery, so it lands
|
||||
/// before the replay rather than after RefreshEnd.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public async Task A_folder_refresh_waits_for_a_deferred_delivery_already_past_its_check()
|
||||
{
|
||||
var ct = TestContext.Current.CancellationToken;
|
||||
await using var rig = await ServerEventRig.StartAsync("Refresh855Gate", encrypted: false, ct);
|
||||
rig.KeyRoles = [];
|
||||
rig.MaterialiseNative();
|
||||
rig.Enforce(false, [ServerEventRig.Key()], KeyEquipmentGrant(EqA));
|
||||
|
||||
using var session = await rig.ConnectAsync(anonymous: false, ct);
|
||||
var events = new RecordedEvents();
|
||||
var subscription = await RecordedEvents.SubscribeAsync(session, events, ct);
|
||||
var onA = RecordedEvents.AddEventItem(subscription, ServerEventRig.UnsNodeId(session, EqA));
|
||||
await subscription.ApplyChangesAsync(ct);
|
||||
|
||||
rig.Project(native: true, Active("seed"));
|
||||
await RecordedEvents.WaitUntilAsync(() => events.Has(onA, "seed"));
|
||||
|
||||
using var entered = new ManualResetEventSlim();
|
||||
using var proceed = new ManualResetEventSlim();
|
||||
var probed = 0;
|
||||
rig.Nm.DeferredFolderDeliveryProbe = () =>
|
||||
{
|
||||
if (Interlocked.Increment(ref probed) != 1) return;
|
||||
entered.Set();
|
||||
proceed.Wait(TimeSpan.FromSeconds(30));
|
||||
};
|
||||
|
||||
using (NotifierStall.Hold(rig.Nm, EqA))
|
||||
{
|
||||
for (var i = 0; i < FloodSize; i++) rig.Project(native: true, Active($"f{i}"));
|
||||
rig.Project(native: true, Active("final"));
|
||||
}
|
||||
|
||||
// The released walk parks on its first deferred delivery to the folder item, past the check, inside the gate.
|
||||
entered.Wait(TimeSpan.FromSeconds(10), ct).ShouldBeTrue("the drain must reach the folder item");
|
||||
var refresh = session.CallAsync(
|
||||
ObjectTypeIds.ConditionType, MethodIds.ConditionType_ConditionRefresh, ct, subscription.Id);
|
||||
await Task.Delay(TimeSpan.FromSeconds(1), ct);
|
||||
events.HasType(onA, 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(() => events.HasType(onA, ObjectTypeIds.RefreshEndEventType));
|
||||
rig.Nm.DeferredFolderDeliveryProbe = null;
|
||||
|
||||
await RecordedEvents.WaitUntilAsync(() => rig.Nm.DeferredEventReports == 0);
|
||||
rig.Project(native: true, Active("after"));
|
||||
await RecordedEvents.WaitUntilAsync(() => events.Has(onA, "after"));
|
||||
await Task.Delay(TimeSpan.FromMilliseconds(300), ct);
|
||||
|
||||
var (window, afterEnd) = Window(events.Of(onA));
|
||||
window.Count.ShouldBe(2, "the parked delivery, then the replay");
|
||||
window[0]!.ShouldStartWith("f");
|
||||
window[1].ShouldBe("final");
|
||||
afterEnd.ShouldBe(["after"]);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// The cut path queues its frozen replay OFF <c>Lock</c>. An event newer than the cut that reaches the folder item
|
||||
/// in that window (a condition delivered synchronously, 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_folder_cut_is_never_followed_by_its_older_frozen_replay()
|
||||
{
|
||||
var ct = TestContext.Current.CancellationToken;
|
||||
await using var rig = await ServerEventRig.StartAsync("Refresh855Pending", encrypted: false, ct);
|
||||
rig.KeyRoles = [];
|
||||
rig.MaterialiseNative();
|
||||
rig.Enforce(false, [ServerEventRig.Key()], KeyEquipmentGrant(EqA));
|
||||
// A second condition under the device folder only: its walk never reaches the stalled EqA, so it stays 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);
|
||||
|
||||
// EqA needs an event item of its own to be a notifier fan-out the stall can apply to.
|
||||
using var eqASession = await rig.ConnectAsync(anonymous: false, ct);
|
||||
var eqAEvents = new RecordedEvents();
|
||||
var eqASubscription = await RecordedEvents.SubscribeAsync(eqASession, eqAEvents, ct);
|
||||
RecordedEvents.AddEventItem(eqASubscription, ServerEventRig.UnsNodeId(eqASession, EqA));
|
||||
await eqASubscription.ApplyChangesAsync(ct);
|
||||
|
||||
// The refreshed item is on the raw device folder, which both conditions report through.
|
||||
using var session = await rig.ConnectAsync(anonymous: false, ct);
|
||||
var events = new RecordedEvents();
|
||||
var subscription = await RecordedEvents.SubscribeAsync(session, events, ct);
|
||||
var onDevice = RecordedEvents.AddEventItem(subscription, RawNodeId(session, ServerEventRig.RawDeviceFolder));
|
||||
await subscription.ApplyChangesAsync(ct);
|
||||
onDevice.Status.Error.ShouldBeNull();
|
||||
|
||||
rig.Project(native: true, Active("seed"));
|
||||
await RecordedEvents.WaitUntilAsync(() => events.Has(onDevice, "seed"));
|
||||
|
||||
var injected = 0;
|
||||
rig.Nm.DirectRefreshProbe = (cut, _) =>
|
||||
{
|
||||
if (cut && Interlocked.Exchange(ref injected, 1) == 0)
|
||||
rig.Sink.WriteAlarmCondition(Free, Active("x2"), DateTime.UtcNow, AddressSpaceRealm.Raw);
|
||||
};
|
||||
|
||||
using (var stall = NotifierStall.Hold(rig.Nm, EqA))
|
||||
{
|
||||
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 session.CallAsync(ObjectTypeIds.ConditionType, MethodIds.ConditionType_ConditionRefresh, ct, subscription.Id);
|
||||
await RecordedEvents.WaitUntilAsync(() => events.HasType(onDevice, ObjectTypeIds.RefreshEndEventType));
|
||||
stall.Release();
|
||||
}
|
||||
|
||||
rig.Nm.DirectRefreshProbe = null;
|
||||
Volatile.Read(ref injected).ShouldBe(1, "the folder refresh took the cut path");
|
||||
var stream = events.Of(onDevice).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>
|
||||
/// ACL on: a condition the folder refresh omitted because the item may not see it there (its reference into EqA was
|
||||
/// dropped by the generation) is not withheld afterwards — the reference returns before the drain, and the item must
|
||||
/// end on the condition's newest state. And while omitted, the replay itself carried nothing of it.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public async Task Deferred_events_of_a_condition_the_folder_refresh_omitted_by_the_acl_are_not_withheld()
|
||||
{
|
||||
var ct = TestContext.Current.CancellationToken;
|
||||
await using var rig = await ServerEventRig.StartAsync("Refresh855Acl", encrypted: false, ct);
|
||||
rig.KeyRoles = [];
|
||||
rig.MaterialiseNative();
|
||||
NodeAcl[] grants = [KeyEquipmentGrant(EqA)];
|
||||
var state = rig.Enforce(true, [ServerEventRig.Key()], grants);
|
||||
Deploy(state, "rev-2", grants, ServerEventRig.EqB); // the native alarm is no longer projected into EqA
|
||||
|
||||
using var session = await rig.ConnectAsync(anonymous: false, ct);
|
||||
var events = new RecordedEvents();
|
||||
var subscription = await RecordedEvents.SubscribeAsync(session, events, ct);
|
||||
var onA = RecordedEvents.AddEventItem(subscription, ServerEventRig.UnsNodeId(session, EqA));
|
||||
await subscription.ApplyChangesAsync(ct);
|
||||
onA.Status.Error.ShouldBeNull();
|
||||
|
||||
rig.Project(native: true, Active("seed"));
|
||||
await Task.Delay(TimeSpan.FromMilliseconds(500), ct);
|
||||
events.Has(onA, "seed").ShouldBeFalse("the generation does not project the alarm into EqA");
|
||||
|
||||
using (var stall = NotifierStall.Hold(rig.Nm, EqA))
|
||||
{
|
||||
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 session.CallAsync(ObjectTypeIds.ConditionType, MethodIds.ConditionType_ConditionRefresh, ct, subscription.Id);
|
||||
await RecordedEvents.WaitUntilAsync(() => events.HasType(onA, ObjectTypeIds.RefreshEndEventType));
|
||||
|
||||
// The reference returns while the condition's events still wait.
|
||||
Deploy(state, "rev-3", grants, EqA, ServerEventRig.EqB);
|
||||
stall.Release();
|
||||
}
|
||||
|
||||
await RecordedEvents.WaitUntilAsync(() => rig.Nm.DeferredEventReports == 0);
|
||||
await RecordedEvents.WaitUntilAsync(() => events.Has(onA, "final"));
|
||||
var (replayed, after) = Window(events.Of(onA));
|
||||
replayed.ShouldBeEmpty("the refresh omitted the condition: not visible through EqA at the time");
|
||||
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')).ShouldBeGreaterThan(0, "its deferred events were delivered, not withheld");
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// ACL on: a grant revoked while a refreshed folder item's deferred events wait. Neither the replay (taken while
|
||||
/// granted) nor anything after it leaks once revoked; a still-granted subscriber on the same folder gets the tail.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public async Task A_grant_revoked_after_a_folder_refresh_withholds_every_deferred_event()
|
||||
{
|
||||
var ct = TestContext.Current.CancellationToken;
|
||||
await using var rig = await ServerEventRig.StartAsync("Refresh855Revoke", encrypted: false, ct);
|
||||
rig.KeyRoles = [];
|
||||
rig.MaterialiseNative();
|
||||
var anonGrant = ServerEventRig.EquipmentGrant(EqA, NodePermissions.ReadOnly);
|
||||
var state = rig.Enforce(true, [ServerEventRig.Key()], KeyEquipmentGrant(EqA), anonGrant);
|
||||
|
||||
using var keyed = await rig.ConnectAsync(anonymous: false, ct);
|
||||
var keyEvents = new RecordedEvents();
|
||||
var keySub = await RecordedEvents.SubscribeAsync(keyed, keyEvents, ct);
|
||||
var keyA = RecordedEvents.AddEventItem(keySub, ServerEventRig.UnsNodeId(keyed, EqA));
|
||||
await keySub.ApplyChangesAsync(ct);
|
||||
|
||||
using var anonymous = await rig.ConnectAsync(anonymous: true, ct);
|
||||
var anonEvents = new RecordedEvents();
|
||||
var anonSub = await RecordedEvents.SubscribeAsync(anonymous, anonEvents, ct);
|
||||
var anonA = RecordedEvents.AddEventItem(anonSub, ServerEventRig.UnsNodeId(anonymous, EqA));
|
||||
await anonSub.ApplyChangesAsync(ct);
|
||||
|
||||
rig.Project(native: true, Active("seed"));
|
||||
await RecordedEvents.WaitUntilAsync(() => keyEvents.Has(keyA, "seed") && anonEvents.Has(anonA, "seed"));
|
||||
|
||||
using (var stall = NotifierStall.Hold(rig.Nm, EqA))
|
||||
{
|
||||
for (var i = 0; i < FloodSize; i++) rig.Project(native: true, Active($"f{i}"));
|
||||
await keyed.CallAsync(ObjectTypeIds.ConditionType, MethodIds.ConditionType_ConditionRefresh, ct, keySub.Id);
|
||||
await RecordedEvents.WaitUntilAsync(() => keyEvents.HasType(keyA, ObjectTypeIds.RefreshEndEventType));
|
||||
|
||||
state.Apply(
|
||||
ServerEventRig.Cluster, "rev-2", [ServerEventRig.Key()], [anonGrant], ServerEventRig.ScopePaths(),
|
||||
ServerEventRig.EquipmentOwnerPaths(), ServerEventRig.ScriptedAlarmOwners(), ServerEventRig.FolderScopes(),
|
||||
ServerEventRig.TagReferences());
|
||||
rig.Project(native: true, Active("after-revoke"));
|
||||
stall.Release();
|
||||
}
|
||||
|
||||
await RecordedEvents.WaitUntilAsync(() => rig.Nm.DeferredEventReports == 0);
|
||||
await RecordedEvents.WaitUntilAsync(() =>
|
||||
anonEvents.Has(anonA, $"f{FloodSize - 1}") && anonEvents.Has(anonA, "after-revoke"));
|
||||
await Task.Delay(TimeSpan.FromSeconds(1), ct);
|
||||
|
||||
var (_, after) = Window(keyEvents.Of(keyA));
|
||||
after.ShouldBeEmpty("after RefreshEnd the revoked session gets nothing: older events are superseded, later ones denied");
|
||||
keyEvents.Has(keyA, "after-revoke").ShouldBeFalse();
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// #855 review: the cut path decides its replay before taking <c>Lock</c>; a grant revoked between that decision
|
||||
/// and the queueing of the frozen replay must still withhold the replay (it is re-decided right before each entry).
|
||||
/// The revocation lands from the refresh probe, which runs after the cut is installed and before the replay is queued.
|
||||
/// </summary>
|
||||
/// <param name="serverObject">Refresh a Server-object item instead of a folder item.</param>
|
||||
[Theory]
|
||||
[InlineData(false)]
|
||||
[InlineData(true)]
|
||||
public async Task A_grant_revoked_between_the_cut_and_its_replay_withholds_the_replay(bool serverObject)
|
||||
{
|
||||
var ct = TestContext.Current.CancellationToken;
|
||||
await using var rig = await ServerEventRig.StartAsync("Refresh855RevokeCut", encrypted: false, ct);
|
||||
rig.KeyRoles = [];
|
||||
rig.MaterialiseNative();
|
||||
var anonGrant = ServerEventRig.EquipmentGrant(EqA, NodePermissions.ReadOnly);
|
||||
var state = rig.Enforce(true, [ServerEventRig.Key()], KeyEquipmentGrant(EqA), anonGrant);
|
||||
|
||||
// An anonymous subscriber keeps EqA a notifier fan-out the stall applies to (and stays granted throughout).
|
||||
using var anonymous = await rig.ConnectAsync(anonymous: true, ct);
|
||||
var anonEvents = new RecordedEvents();
|
||||
var anonSub = await RecordedEvents.SubscribeAsync(anonymous, anonEvents, ct);
|
||||
RecordedEvents.AddEventItem(anonSub, ServerEventRig.UnsNodeId(anonymous, EqA));
|
||||
await anonSub.ApplyChangesAsync(ct);
|
||||
|
||||
using var keyed = await rig.ConnectAsync(anonymous: false, ct);
|
||||
var keyEvents = new RecordedEvents();
|
||||
var keySub = await RecordedEvents.SubscribeAsync(keyed, keyEvents, ct);
|
||||
var item = RecordedEvents.AddEventItem(keySub, serverObject ? ObjectIds.Server : ServerEventRig.UnsNodeId(keyed, EqA));
|
||||
await keySub.ApplyChangesAsync(ct);
|
||||
item.Status.Error.ShouldBeNull();
|
||||
|
||||
rig.Project(native: true, Active("seed"));
|
||||
await RecordedEvents.WaitUntilAsync(() => keyEvents.Has(item, "seed"));
|
||||
|
||||
var revoked = 0;
|
||||
void Revoke(bool cut, bool _)
|
||||
{
|
||||
if (!cut || Interlocked.Exchange(ref revoked, 1) != 0) return;
|
||||
state.Apply(
|
||||
ServerEventRig.Cluster, "rev-2", [ServerEventRig.Key()], [anonGrant], ServerEventRig.ScopePaths(),
|
||||
ServerEventRig.EquipmentOwnerPaths(), ServerEventRig.ScriptedAlarmOwners(), ServerEventRig.FolderScopes(),
|
||||
ServerEventRig.TagReferences());
|
||||
}
|
||||
|
||||
if (serverObject) rig.Nm.ServerObjectRefreshProbe = Revoke;
|
||||
else rig.Nm.DirectRefreshProbe = Revoke;
|
||||
|
||||
using (var stall = NotifierStall.Hold(rig.Nm, EqA))
|
||||
{
|
||||
for (var i = 0; i < FloodSize; i++) rig.Project(native: true, Active($"f{i}"));
|
||||
rig.Nm.DeferredEventReports.ShouldBeGreaterThan(0);
|
||||
await keyed.CallAsync(ObjectTypeIds.ConditionType, MethodIds.ConditionType_ConditionRefresh, ct, keySub.Id);
|
||||
await RecordedEvents.WaitUntilAsync(() => keyEvents.HasType(item, ObjectTypeIds.RefreshEndEventType));
|
||||
stall.Release();
|
||||
}
|
||||
|
||||
rig.Nm.ServerObjectRefreshProbe = null;
|
||||
rig.Nm.DirectRefreshProbe = null;
|
||||
Volatile.Read(ref revoked).ShouldBe(1, "the refresh took the cut path");
|
||||
await RecordedEvents.WaitUntilAsync(() => rig.Nm.DeferredEventReports == 0);
|
||||
await Task.Delay(TimeSpan.FromMilliseconds(500), ct);
|
||||
|
||||
var (replayed, after) = Window(keyEvents.Of(item));
|
||||
replayed.ShouldBeEmpty("the grant was revoked before the replay was queued");
|
||||
after.ShouldBeEmpty("and nothing reaches the item after RefreshEnd either");
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// #855 review 2: a condition removed by a deploy between the cut and the queueing of its frozen replay must not be
|
||||
/// replayed — its frozen Retain=true state would be a ghost no Retain=false ever follows. ACL off, so nothing but the
|
||||
/// liveness check stands in the way.
|
||||
/// </summary>
|
||||
/// <param name="serverObject">Refresh a Server-object item instead of a folder item.</param>
|
||||
[Theory]
|
||||
[InlineData(false)]
|
||||
[InlineData(true)]
|
||||
public async Task A_condition_removed_between_the_cut_and_its_replay_is_not_replayed(bool serverObject)
|
||||
{
|
||||
var ct = TestContext.Current.CancellationToken;
|
||||
await using var rig = await ServerEventRig.StartAsync("Refresh855RemoveCut", encrypted: false, ct);
|
||||
rig.KeyRoles = [];
|
||||
rig.MaterialiseNative();
|
||||
rig.Enforce(false, [ServerEventRig.Key()], KeyEquipmentGrant(EqA));
|
||||
|
||||
// A second subscriber keeps EqA a notifier fan-out the stall applies to.
|
||||
using var other = await rig.ConnectAsync(anonymous: true, ct);
|
||||
var otherEvents = new RecordedEvents();
|
||||
var otherSub = await RecordedEvents.SubscribeAsync(other, otherEvents, ct);
|
||||
RecordedEvents.AddEventItem(otherSub, ServerEventRig.UnsNodeId(other, EqA));
|
||||
await otherSub.ApplyChangesAsync(ct);
|
||||
|
||||
using var session = await rig.ConnectAsync(anonymous: false, ct);
|
||||
var events = new RecordedEvents();
|
||||
var subscription = await RecordedEvents.SubscribeAsync(session, events, ct);
|
||||
var item = RecordedEvents.AddEventItem(subscription, serverObject ? ObjectIds.Server : ServerEventRig.UnsNodeId(session, EqA));
|
||||
await subscription.ApplyChangesAsync(ct);
|
||||
|
||||
rig.Project(native: true, Active("seed"));
|
||||
await RecordedEvents.WaitUntilAsync(() => events.Has(item, "seed"));
|
||||
|
||||
var removed = 0;
|
||||
void Remove(bool cut, bool _)
|
||||
{
|
||||
if (cut && Interlocked.Exchange(ref removed, 1) == 0)
|
||||
rig.Nm.RemoveAlarmConditionNode(ServerEventRig.RawAlarmPath, AddressSpaceRealm.Raw).ShouldBeTrue();
|
||||
}
|
||||
|
||||
if (serverObject) rig.Nm.ServerObjectRefreshProbe = Remove;
|
||||
else rig.Nm.DirectRefreshProbe = Remove;
|
||||
|
||||
using (var stall = NotifierStall.Hold(rig.Nm, EqA))
|
||||
{
|
||||
for (var i = 0; i < FloodSize; i++) rig.Project(native: true, Active($"f{i}"));
|
||||
rig.Nm.DeferredEventReports.ShouldBeGreaterThan(0);
|
||||
await session.CallAsync(ObjectTypeIds.ConditionType, MethodIds.ConditionType_ConditionRefresh, ct, subscription.Id);
|
||||
await RecordedEvents.WaitUntilAsync(() => events.HasType(item, ObjectTypeIds.RefreshEndEventType));
|
||||
stall.Release();
|
||||
}
|
||||
|
||||
rig.Nm.ServerObjectRefreshProbe = null;
|
||||
rig.Nm.DirectRefreshProbe = null;
|
||||
Volatile.Read(ref removed).ShouldBe(1, "the refresh took the cut path");
|
||||
await RecordedEvents.WaitUntilAsync(() => rig.Nm.DeferredEventReports == 0);
|
||||
await Task.Delay(TimeSpan.FromMilliseconds(500), ct);
|
||||
|
||||
var (replayed, after) = Window(events.Of(item));
|
||||
replayed.ShouldBeEmpty("the condition was removed before its frozen replay was queued");
|
||||
after.ShouldBeEmpty("and none of its deferred events follows");
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// #855 review 3: the plain refresh path (nothing deferred — the only path production takes) collects live
|
||||
/// conditions off <c>Lock</c> and queues them afterwards; a condition removed in between must not be replayed.
|
||||
/// </summary>
|
||||
/// <param name="serverObject">Refresh a Server-object item instead of a folder item.</param>
|
||||
[Theory]
|
||||
[InlineData(false)]
|
||||
[InlineData(true)]
|
||||
public async Task A_condition_removed_during_a_plain_refresh_is_not_replayed(bool serverObject)
|
||||
{
|
||||
var ct = TestContext.Current.CancellationToken;
|
||||
await using var rig = await ServerEventRig.StartAsync("Refresh855RemovePlain", encrypted: false, ct);
|
||||
rig.KeyRoles = [];
|
||||
rig.MaterialiseNative();
|
||||
rig.Enforce(false, [ServerEventRig.Key()], KeyEquipmentGrant(EqA));
|
||||
|
||||
using var session = await rig.ConnectAsync(anonymous: false, ct);
|
||||
var events = new RecordedEvents();
|
||||
var subscription = await RecordedEvents.SubscribeAsync(session, events, ct);
|
||||
var item = RecordedEvents.AddEventItem(subscription, serverObject ? ObjectIds.Server : ServerEventRig.UnsNodeId(session, EqA));
|
||||
await subscription.ApplyChangesAsync(ct);
|
||||
|
||||
rig.Project(native: true, Active("seed"));
|
||||
await RecordedEvents.WaitUntilAsync(() => events.Has(item, "seed"));
|
||||
|
||||
var removed = 0;
|
||||
void Remove(bool cut, bool _)
|
||||
{
|
||||
if (!cut && Interlocked.Exchange(ref removed, 1) == 0)
|
||||
rig.Nm.RemoveAlarmConditionNode(ServerEventRig.RawAlarmPath, AddressSpaceRealm.Raw).ShouldBeTrue();
|
||||
}
|
||||
|
||||
if (serverObject) rig.Nm.ServerObjectRefreshProbe = Remove;
|
||||
else rig.Nm.DirectRefreshProbe = Remove;
|
||||
await session.CallAsync(ObjectTypeIds.ConditionType, MethodIds.ConditionType_ConditionRefresh, ct, subscription.Id);
|
||||
await RecordedEvents.WaitUntilAsync(() => events.HasType(item, ObjectTypeIds.RefreshEndEventType));
|
||||
rig.Nm.ServerObjectRefreshProbe = null;
|
||||
rig.Nm.DirectRefreshProbe = null;
|
||||
|
||||
Volatile.Read(ref removed).ShouldBe(1, "the plain path ran");
|
||||
Window(events.Of(item)).Replayed.ShouldBeEmpty("the condition was removed before the replay was queued");
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// #855 review: a delivery that fails inline (here: a hop that throws) is over, not suspended — it must not be booked
|
||||
/// as a deferral episode (which moved the episode a concurrent plain refresh checks, and cleared cuts under the
|
||||
/// event-report gate while <c>Lock</c> was held). The Server-object item, served before the failing hop, still gets
|
||||
/// the event, and the next event is delivered normally.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public async Task A_delivery_that_throws_is_not_booked_as_a_deferral()
|
||||
{
|
||||
var ct = TestContext.Current.CancellationToken;
|
||||
await using var rig = await ServerEventRig.StartAsync("Refresh855Fault", encrypted: false, ct);
|
||||
rig.KeyRoles = [];
|
||||
rig.MaterialiseNative();
|
||||
rig.Enforce(false, [ServerEventRig.Key()], KeyEquipmentGrant(EqA));
|
||||
|
||||
using var session = await rig.ConnectAsync(anonymous: false, ct);
|
||||
var events = new RecordedEvents();
|
||||
var subscription = await RecordedEvents.SubscribeAsync(session, events, ct);
|
||||
var onA = RecordedEvents.AddEventItem(subscription, ServerEventRig.UnsNodeId(session, EqA));
|
||||
var onServer = RecordedEvents.AddEventItem(subscription, ObjectIds.Server);
|
||||
await subscription.ApplyChangesAsync(ct);
|
||||
|
||||
var episodes = rig.Nm.DeferralEpisodes;
|
||||
rig.Nm.NotifierDeliveryStall = node =>
|
||||
node.NodeId.IdentifierAsString == EqA ? throw new InvalidOperationException("hop fault (test)") : null;
|
||||
rig.Project(native: true, Active("faulted"));
|
||||
rig.Nm.NotifierDeliveryStall = null;
|
||||
|
||||
rig.Nm.DeferredEventReports.ShouldBe(0);
|
||||
rig.Nm.DeferralEpisodes.ShouldBe(episodes, "a failed delivery is not a deferral episode");
|
||||
rig.Project(native: true, Active("next"));
|
||||
await RecordedEvents.WaitUntilAsync(() =>
|
||||
events.Has(onServer, "faulted") && events.Has(onServer, "next") && events.Has(onA, "next"));
|
||||
events.Has(onA, "faulted").ShouldBeFalse("the faulted hop delivered nothing");
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Production shape, no stall: bursts reported under one <c>Lock</c> hold (an apply pass) and single reports, against
|
||||
/// folder items that refresh repeatedly while their client is slow to publish. Folder items are served per item, so
|
||||
/// nothing is ever deferred; each refresh's replay is never older than an event the item already received, nothing
|
||||
/// after RefreshEnd goes backwards, the stream converges on the newest state, and no replay is queued under <c>Lock</c>.
|
||||
/// </summary>
|
||||
/// <param name="enforced">Per-folder ACL on or off.</param>
|
||||
[Theory]
|
||||
[InlineData(false)]
|
||||
[InlineData(true)]
|
||||
public async Task Production_overload_never_defers_and_never_overtakes_a_folder_refresh(bool enforced)
|
||||
{
|
||||
var ct = TestContext.Current.CancellationToken;
|
||||
await using var rig = await ServerEventRig.StartAsync("Refresh855Load", encrypted: false, ct);
|
||||
rig.KeyRoles = [];
|
||||
rig.MaterialiseNative();
|
||||
rig.Enforce(enforced, [ServerEventRig.Key()], KeyEquipmentGrant(EqA));
|
||||
|
||||
using var session = await rig.ConnectAsync(anonymous: false, ct);
|
||||
var events = new RecordedEvents();
|
||||
var subscription = await RecordedEvents.SubscribeAsync(session, events, ct);
|
||||
var onA = RecordedEvents.AddEventItem(subscription, ServerEventRig.UnsNodeId(session, EqA));
|
||||
onA.QueueSize = 100_000; // keep every event so the ordering check sees the whole stream
|
||||
await subscription.ApplyChangesAsync(ct);
|
||||
|
||||
rig.Project(native: true, Active("s0"));
|
||||
await RecordedEvents.WaitUntilAsync(() => events.Has(onA, "s0"));
|
||||
|
||||
var probes = new System.Collections.Concurrent.ConcurrentQueue<(bool Cut, bool LockHeld)>();
|
||||
rig.Nm.DirectRefreshProbe = (cut, held) => probes.Enqueue((cut, held));
|
||||
|
||||
using var stop = CancellationTokenSource.CreateLinkedTokenSource(ct);
|
||||
var counter = 0;
|
||||
var peakDeferred = 0;
|
||||
var reporter = Task.Factory.StartNew(() =>
|
||||
{
|
||||
while (!stop.IsCancellationRequested)
|
||||
{
|
||||
lock (rig.Nm.Lock)
|
||||
{
|
||||
for (var i = 0; i < 1000; i++)
|
||||
{
|
||||
rig.Project(native: true, Active($"s{Interlocked.Increment(ref counter)}"));
|
||||
// Sampled inside Lock, after each report: a deferral cannot open and drain unseen (#855 review).
|
||||
peakDeferred = Math.Max(peakDeferred, rig.Nm.DeferredEventReports);
|
||||
}
|
||||
}
|
||||
|
||||
rig.Project(native: true, Active($"s{Interlocked.Increment(ref counter)}"));
|
||||
Thread.Sleep(2);
|
||||
}
|
||||
}, stop.Token, TaskCreationOptions.LongRunning, TaskScheduler.Default);
|
||||
|
||||
var refusedAsInProgress = 0;
|
||||
for (var r = 0; r < 30; r++)
|
||||
{
|
||||
try
|
||||
{
|
||||
await RefreshAsync(session, subscription, onA, refresh2: r % 3 == 2, ct).WaitAsync(TimeSpan.FromSeconds(20), ct);
|
||||
}
|
||||
catch (ServiceResultException ex) when (ex.StatusCode == StatusCodes.BadRefreshInProgress)
|
||||
{
|
||||
refusedAsInProgress++;
|
||||
}
|
||||
|
||||
await Task.Delay(TimeSpan.FromMilliseconds(80), ct);
|
||||
}
|
||||
|
||||
await stop.CancelAsync();
|
||||
await reporter.WaitAsync(TimeSpan.FromSeconds(30), CancellationToken.None);
|
||||
rig.Nm.DirectRefreshProbe = null;
|
||||
|
||||
peakDeferred.ShouldBe(0, "no notifier of ours defers: folder items are served per item by the reporter");
|
||||
rig.Nm.DiscardedEventReports.ShouldBe(0);
|
||||
var newest = $"s{Volatile.Read(ref counter)}";
|
||||
await RecordedEvents.WaitUntilAsync(() => events.Has(onA, newest));
|
||||
await Task.Delay(TimeSpan.FromMilliseconds(500), ct);
|
||||
|
||||
CheckStream(events.Of(onA)).ShouldBeEmpty();
|
||||
refusedAsInProgress.ShouldBeLessThan(30, "most refreshes must actually have run");
|
||||
probes.ShouldNotBeEmpty();
|
||||
probes.ShouldAllBe(p => !p.Cut && !p.LockHeld);
|
||||
RecordedEvents.Message(events.Of(onA).Last(f => f[RecordedEvents.AckedField].TryGetValue(out bool _)))
|
||||
.ShouldBe(newest, "the stream converges on the newest state");
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Race / stress with deferral: reports (bursts under <c>Lock</c> and single), repeated refreshes of the folder item
|
||||
/// and an intermittent stall of its folder's delivery all run at once. Outside a refresh the folder item's stream
|
||||
/// never goes backwards, each refresh ends on the newest state the item has seen, and it converges. Nothing
|
||||
/// deadlocks, and neither refresh path queues its replay under <c>Lock</c>.
|
||||
/// </summary>
|
||||
/// <param name="enforced">Per-folder ACL on or off.</param>
|
||||
[Theory]
|
||||
[InlineData(false)]
|
||||
[InlineData(true)]
|
||||
public async Task Concurrent_reports_folder_refreshes_and_stalls_never_deliver_a_pre_refresh_state_after_RefreshEnd(bool enforced)
|
||||
{
|
||||
var ct = TestContext.Current.CancellationToken;
|
||||
await using var rig = await ServerEventRig.StartAsync("Refresh855Stress", encrypted: false, ct);
|
||||
rig.KeyRoles = [];
|
||||
rig.MaterialiseNative();
|
||||
rig.Enforce(enforced, [ServerEventRig.Key()], KeyEquipmentGrant(EqA));
|
||||
|
||||
using var session = await rig.ConnectAsync(anonymous: false, ct);
|
||||
var events = new RecordedEvents();
|
||||
var subscription = await RecordedEvents.SubscribeAsync(session, events, ct);
|
||||
var onA = RecordedEvents.AddEventItem(subscription, ServerEventRig.UnsNodeId(session, EqA));
|
||||
onA.QueueSize = 100_000;
|
||||
await subscription.ApplyChangesAsync(ct);
|
||||
|
||||
rig.Project(native: true, Active("s0"));
|
||||
await RecordedEvents.WaitUntilAsync(() => events.Has(onA, "s0"));
|
||||
|
||||
var probes = new System.Collections.Concurrent.ConcurrentQueue<(bool Cut, bool LockHeld)>();
|
||||
rig.Nm.DirectRefreshProbe = (cut, held) => probes.Enqueue((cut, held));
|
||||
|
||||
using var stop = CancellationTokenSource.CreateLinkedTokenSource(ct);
|
||||
var counter = 0;
|
||||
var reporter = Task.Factory.StartNew(() =>
|
||||
{
|
||||
while (!stop.IsCancellationRequested)
|
||||
{
|
||||
lock (rig.Nm.Lock)
|
||||
{
|
||||
for (var i = 0; i < 300; 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 (NotifierStall.Hold(rig.Nm, EqA))
|
||||
{
|
||||
Thread.Sleep(300);
|
||||
if (rig.Nm.DeferredEventReports > 0) Interlocked.Increment(ref deferredSeen);
|
||||
}
|
||||
|
||||
Thread.Sleep(100);
|
||||
}
|
||||
}, stop.Token, TaskCreationOptions.LongRunning, TaskScheduler.Default);
|
||||
|
||||
var refusedAsInProgress = 0;
|
||||
var stressDeadline = DateTime.UtcNow + TimeSpan.FromSeconds(90);
|
||||
for (var r = 0; r < 40 || (Volatile.Read(ref deferredSeen) <= 1 && DateTime.UtcNow < stressDeadline); r++)
|
||||
{
|
||||
try
|
||||
{
|
||||
await RefreshAsync(session, subscription, onA, refresh2: r % 3 == 2, ct).WaitAsync(TimeSpan.FromSeconds(20), ct);
|
||||
}
|
||||
catch (ServiceResultException ex) when (ex.StatusCode == StatusCodes.BadRefreshInProgress)
|
||||
{
|
||||
refusedAsInProgress++;
|
||||
}
|
||||
|
||||
await Task.Delay(TimeSpan.FromMilliseconds(80), ct);
|
||||
}
|
||||
|
||||
await stop.CancelAsync();
|
||||
await reporter.WaitAsync(TimeSpan.FromSeconds(30), CancellationToken.None);
|
||||
await staller.WaitAsync(TimeSpan.FromSeconds(30), CancellationToken.None);
|
||||
rig.Nm.DirectRefreshProbe = null;
|
||||
Volatile.Read(ref deferredSeen).ShouldBeGreaterThan(1, "the stress must actually have run with deferred events");
|
||||
|
||||
await RecordedEvents.WaitUntilAsync(() => rig.Nm.DeferredEventReports == 0);
|
||||
var newest = $"s{Volatile.Read(ref counter)}";
|
||||
await RecordedEvents.WaitUntilAsync(() => events.Has(onA, newest));
|
||||
await Task.Delay(TimeSpan.FromMilliseconds(500), ct);
|
||||
|
||||
CheckStream(events.Of(onA)).ShouldBeEmpty();
|
||||
refusedAsInProgress.ShouldBeLessThan(40, "most refreshes must actually have run");
|
||||
rig.Nm.RefreshSupersededDeliveries.ShouldBeGreaterThan(0, "some folder refresh must actually have overtaken deferred events");
|
||||
probes.ShouldContain(p => p.Cut, "some folder refresh must have taken the cut path");
|
||||
probes.ShouldAllBe(p => !p.LockHeld);
|
||||
RecordedEvents.Message(events.Of(onA).Last(f => f[RecordedEvents.AckedField].TryGetValue(out bool _)))
|
||||
.ShouldBe(newest, "the stream converges on the newest state");
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Walks one condition's stream: 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 follows).
|
||||
/// </summary>
|
||||
private static List<string> CheckStream(IReadOnlyList<ArrayOf<Variant>> stream)
|
||||
{
|
||||
var max = -1;
|
||||
var last = -1;
|
||||
var replaying = false;
|
||||
var violations = new List<string>();
|
||||
foreach (var f in stream)
|
||||
{
|
||||
var type = f[RecordedEvents.EventTypeField].GetNodeId();
|
||||
if (type == ObjectTypeIds.RefreshStartEventType)
|
||||
{
|
||||
replaying = true;
|
||||
continue;
|
||||
}
|
||||
|
||||
if (type == ObjectTypeIds.RefreshEndEventType)
|
||||
{
|
||||
replaying = false;
|
||||
if (last < max) violations.Add($"a refresh ended on s{last} after s{max} had been delivered");
|
||||
continue;
|
||||
}
|
||||
|
||||
if (!f[RecordedEvents.AckedField].TryGetValue(out bool _)) continue;
|
||||
var n = int.Parse(RecordedEvents.Message(f)![1..], CultureInfo.InvariantCulture);
|
||||
if (!replaying && n < max) violations.Add($"s{n} delivered after s{max}");
|
||||
max = Math.Max(max, n);
|
||||
last = n;
|
||||
}
|
||||
|
||||
return violations;
|
||||
}
|
||||
|
||||
private static Task RefreshAsync(ISession session, Subscription subscription, MonitoredItem item, bool refresh2, CancellationToken ct) =>
|
||||
refresh2
|
||||
? session.CallAsync(ObjectTypeIds.ConditionType, MethodIds.ConditionType_ConditionRefresh2, ct, subscription.Id, item.Status.Id)
|
||||
: session.CallAsync(ObjectTypeIds.ConditionType, MethodIds.ConditionType_ConditionRefresh, ct, subscription.Id);
|
||||
|
||||
/// <summary>The condition-event messages inside the item's LAST RefreshStart..RefreshEnd, and those after it.</summary>
|
||||
private static (List<string?> Replayed, List<string?> AfterEnd) Window(IReadOnlyList<ArrayOf<Variant>> stream)
|
||||
{
|
||||
var list = stream.ToList();
|
||||
var start = list.FindLastIndex(f => f[RecordedEvents.EventTypeField].GetNodeId() == ObjectTypeIds.RefreshStartEventType);
|
||||
var end = list.FindLastIndex(f => f[RecordedEvents.EventTypeField].GetNodeId() == ObjectTypeIds.RefreshEndEventType);
|
||||
start.ShouldBeGreaterThanOrEqualTo(0);
|
||||
end.ShouldBeGreaterThan(start);
|
||||
static bool IsCondition(ArrayOf<Variant> f) => f[RecordedEvents.AckedField].TryGetValue(out bool _);
|
||||
return (
|
||||
list.Skip(start + 1).Take(end - start - 1).Where(IsCondition).Select(RecordedEvents.Message).ToList(),
|
||||
list.Skip(end + 1).Where(IsCondition).Select(RecordedEvents.Message).ToList());
|
||||
}
|
||||
|
||||
private static List<int> Indices(IReadOnlyList<ArrayOf<Variant>> received, char prefix) =>
|
||||
received
|
||||
.Select(RecordedEvents.Message)
|
||||
.Where(m => m is { Length: > 1 } && m[0] == prefix && char.IsAsciiDigit(m[1]))
|
||||
.Select(m => int.Parse(m![1..], CultureInfo.InvariantCulture))
|
||||
.ToList();
|
||||
|
||||
private static NodeId RawNodeId(ISession session, string id) =>
|
||||
new(id, (ushort)session.NamespaceUris.GetIndex(V3NodeIds.RawNamespaceUri));
|
||||
|
||||
private static void Deploy(DataPlaneAuthorityState state, string revision, NodeAcl[] grants, params string[] referencing) =>
|
||||
state.Apply(
|
||||
ServerEventRig.Cluster, revision, [ServerEventRig.Key()], grants, ServerEventRig.ScopePaths(),
|
||||
ServerEventRig.EquipmentOwnerPaths(), ServerEventRig.ScriptedAlarmOwners(), ServerEventRig.FolderScopes(),
|
||||
new Dictionary<string, IReadOnlySet<string>> { [ServerEventRig.RawAlarmPath] = referencing.ToHashSet() });
|
||||
|
||||
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,
|
||||
};
|
||||
}
|
||||
+36
-19
@@ -19,6 +19,13 @@ namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests;
|
||||
/// consumer — a deadlock that also freezes every other operation needing <c>Lock</c>. Publishing is switched off
|
||||
/// on the client so nothing downstream drains either (the slow / absent consumer case).
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// Since Gitea #855 condition events no longer go through that channel at all: the node manager queues them to a
|
||||
/// folder's items itself, per item, on the reporter's thread (as it always did for Server-object items), so the same
|
||||
/// flood now completes without deferring anything. The deadlock test therefore also pins that per-item delivery is in
|
||||
/// force (it would defer again if the bypass stopped recognising the SDK's fan-out); the ghost test produces its
|
||||
/// deferral with <see cref="NotifierStall"/>.
|
||||
/// </remarks>
|
||||
public sealed class FolderEventFloodWireTests
|
||||
{
|
||||
private const string EqA = ServerEventRig.EqA;
|
||||
@@ -56,8 +63,12 @@ public sealed class FolderEventFloodWireTests
|
||||
// No publishing: the server-side item queue (QueueSize 1000) is the only sink, as with a stalled client.
|
||||
await subscription.SetPublishingModeAsync(false, ct);
|
||||
|
||||
// #855: the folder's SDK fan-out is the one the node manager recognises and serves per item itself.
|
||||
rig.Nm.TryGetFolder(EqA)!.OnReportEventAsync?.Target.ShouldBeOfType<Opc.Ua.Server.MonitoredNode2>();
|
||||
|
||||
// The whole burst under ONE hold of the node manager's Lock — the shape of any multi-event apply/projection
|
||||
// pass — starting from an uncached verdict (the item is new), so the delivery task needs Lock for event 1.
|
||||
var peakDeferred = 0;
|
||||
var flood = Task.Factory.StartNew(() =>
|
||||
{
|
||||
lock (rig.Nm.Lock)
|
||||
@@ -65,6 +76,7 @@ public sealed class FolderEventFloodWireTests
|
||||
for (var i = 0; i < FloodSize; i++)
|
||||
{
|
||||
rig.Project(native: true, State($"f{i}"));
|
||||
peakDeferred = Math.Max(peakDeferred, rig.Nm.DeferredEventReports);
|
||||
Volatile.Write(ref reported, i + 1);
|
||||
}
|
||||
}
|
||||
@@ -83,11 +95,9 @@ public sealed class FolderEventFloodWireTests
|
||||
+ "on Lock (deadlock)");
|
||||
}
|
||||
|
||||
// The overflow was deferred rather than blocking, and drains once Lock is free.
|
||||
await RecordedEvents.WaitUntilAsync(() => rig.Nm.DeferredEventReports == 0);
|
||||
// …and the SDK's channel (up to 4096 accepted events) finishes draining into the item queue, still with
|
||||
// publishing off, so the count below measures the item queue and not events in transit.
|
||||
await Task.Delay(TimeSpan.FromSeconds(2), ct);
|
||||
// #855: nothing was deferred at all — every event was queued to the folder item by the reporter itself.
|
||||
peakDeferred.ShouldBe(0, "no notifier of ours defers: folder items are served per item, not through the SDK channel");
|
||||
rig.Nm.DeferredEventReports.ShouldBe(0);
|
||||
|
||||
// Delivery is bounded by the item's queue (QueueSize 1000, oldest discarded), not by the flood, and the
|
||||
// events that survive arrive in report order, ending with the last one reported.
|
||||
@@ -104,8 +114,7 @@ public sealed class FolderEventFloodWireTests
|
||||
flooded.Count.ShouldBeLessThanOrEqualTo((int)onA.Status.QueueSize, "delivery is bounded by the item's queue, not the flood");
|
||||
flooded.ShouldBe(flooded.Order().ToList(), "events leave in the order they were reported");
|
||||
flooded[^1].ShouldBe(FloodSize - 1);
|
||||
rig.Nm.DiscardedEventReports.ShouldBeGreaterThan(0,
|
||||
"past the per-condition budget the oldest waiting events give way to the newest (coalesced)");
|
||||
rig.Nm.DiscardedEventReports.ShouldBe(0, "nothing was deferred, so nothing was coalesced (the item queue is the bound)");
|
||||
|
||||
// And the node manager is live: a later event is decided by the per-folder ACL and delivered.
|
||||
rig.Project(native: true, State("after"));
|
||||
@@ -127,7 +136,8 @@ public sealed class FolderEventFloodWireTests
|
||||
/// #838 review — events deferred behind a full channel must not outlive their condition. A deploy that removes
|
||||
/// the condition while its events wait used to let them reach Server-object (<c>i=2253</c>) subscribers after the
|
||||
/// NodeDeleted — a ghost Retain alarm for ScadaBridge — and the folder subscribers too. Every hop of a deferred
|
||||
/// report now re-checks that the condition is still the live instance.
|
||||
/// report now re-checks that the condition is still the live instance (#855: also after a stall at a hop, before
|
||||
/// the folder's items are served).
|
||||
/// </summary>
|
||||
/// <param name="enforced">Per-folder / Server-object ACL on or off (off is the production default).</param>
|
||||
[Theory]
|
||||
@@ -151,16 +161,20 @@ public sealed class FolderEventFloodWireTests
|
||||
onServer.Status.Error.ShouldBeNull();
|
||||
await subscription.SetPublishingModeAsync(false, ct);
|
||||
|
||||
// One Lock hold: flood past the channel (the tail is deferred), then the deploy removes the condition.
|
||||
await Task.Factory.StartNew(() =>
|
||||
// One Lock hold, delivery stalled at the folder: flood (the tail is deferred), then the deploy removes the condition.
|
||||
using (var stall = NotifierStall.Hold(rig.Nm, EqA))
|
||||
{
|
||||
lock (rig.Nm.Lock)
|
||||
await Task.Factory.StartNew(() =>
|
||||
{
|
||||
for (var i = 0; i < FloodSize; i++) rig.Project(native: true, State($"f{i}"));
|
||||
rig.Nm.DeferredEventReports.ShouldBeGreaterThan(0, "the flood must actually have deferred events");
|
||||
rig.Nm.RemoveAlarmConditionNode(ServerEventRig.RawAlarmPath, AddressSpaceRealm.Raw).ShouldBeTrue();
|
||||
}
|
||||
}, ct, TaskCreationOptions.LongRunning, TaskScheduler.Default).WaitAsync(TimeSpan.FromSeconds(60), ct);
|
||||
lock (rig.Nm.Lock)
|
||||
{
|
||||
for (var i = 0; i < FloodSize; i++) rig.Project(native: true, State($"f{i}"));
|
||||
rig.Nm.DeferredEventReports.ShouldBeGreaterThan(0, "the flood must actually have deferred events");
|
||||
rig.Nm.RemoveAlarmConditionNode(ServerEventRig.RawAlarmPath, AddressSpaceRealm.Raw).ShouldBeTrue();
|
||||
}
|
||||
}, ct, TaskCreationOptions.LongRunning, TaskScheduler.Default).WaitAsync(TimeSpan.FromSeconds(60), ct);
|
||||
stall.Release();
|
||||
}
|
||||
|
||||
await RecordedEvents.WaitUntilAsync(() => rig.Nm.DeferredEventReports == 0);
|
||||
rig.Nm.DiscardedEventReports.ShouldBeGreaterThan(0, "the waiting events were dropped, not delivered");
|
||||
@@ -174,12 +188,15 @@ public sealed class FolderEventFloodWireTests
|
||||
.Select(m => int.Parse(m![1..], System.Globalization.CultureInfo.InvariantCulture))
|
||||
.ToArray();
|
||||
|
||||
// Positive control: the events reported before the channel filled did arrive (Server object: synchronously at
|
||||
// the condition's own hop).
|
||||
// Positive control: the event reported before the stall did arrive at the Server object (synchronously, at the
|
||||
// condition's own hop, before the walk reached the stalled folder).
|
||||
Flooded(onServer).ShouldNotBeEmpty();
|
||||
// None of the deferred tail — which was still waiting when the condition was removed — reached anyone.
|
||||
Flooded(onServer).Max().ShouldBeLessThan(FloodSize - 1, "no ghost reached the Server-object subscriber");
|
||||
Flooded(onA).DefaultIfEmpty(-1).Max().ShouldBeLessThan(FloodSize - 1, "no ghost reached the folder subscriber");
|
||||
// Only the in-flight f0 ever reached the stalled EqA hop, and the condition was gone when the stall released: the
|
||||
// post-stall liveness re-check must have withheld it (#855 review 4 — "< FloodSize - 1" passed without that check).
|
||||
Flooded(onA).ShouldBeEmpty("no ghost reached the folder subscriber");
|
||||
Flooded(onServer).ShouldBe([0], "only the event reported before the stall reached the Server object");
|
||||
}
|
||||
|
||||
private static AlarmConditionSnapshot State(string message) =>
|
||||
|
||||
+4
-4
@@ -97,10 +97,10 @@ public sealed class FolderEventVerdictInvalidationTests
|
||||
rig.Project(native: false, State("s1"));
|
||||
await RecordedEvents.WaitUntilAsync(() => events.Has(onA, "s1"));
|
||||
|
||||
// Holding the node-manager lock parks EqA's SDK delivery task: the native event "n1" is queued first and its
|
||||
// verdict (a cache miss after the first invalidation) needs the lock in the inherited role check — its node is
|
||||
// never removed, so that check always takes it. The OLD scripted instance's "s2", queued behind it on the same
|
||||
// single-reader queue, is therefore decided only after the replace and the apply's invalidation.
|
||||
// Before Gitea #855, holding the node-manager lock parked EqA's SDK delivery task, so the OLD scripted instance's
|
||||
// "s2" was decided (and its verdict cached) only after the replace and the apply's invalidation. Condition events
|
||||
// are now decided per item at report time, uncached — "s2" is delivered while its instance is live — and this
|
||||
// pins that the replace still leaves the new instance's events flowing either way.
|
||||
lock (rig.Nm.Lock)
|
||||
{
|
||||
rig.Nm.CompleteAddressSpaceApply(); // forget the cached allows, so both verdicts must be computed
|
||||
|
||||
@@ -0,0 +1,42 @@
|
||||
using Opc.Ua;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests;
|
||||
|
||||
/// <summary>
|
||||
/// Stalls condition-event delivery at one of the node manager's notifiers (Gitea #855): every condition event whose walk
|
||||
/// reaches that notifier suspends there until <see cref="Release"/>, without blocking the reporter or holding the node
|
||||
/// manager's <c>Lock</c> — so the condition's later events are deferred, as they were behind a full SDK channel before
|
||||
/// folder items were delivered per item.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// The pre-#855 tests produced deferral by holding a folder item's server-side <c>MonitoredItem</c> lock, which parked
|
||||
/// the SDK's per-node delivery task. Folder items are now queued synchronously by the reporter (as Server-object items
|
||||
/// always were), so holding an item lock would block the reporter instead; this goes through the node manager's test
|
||||
/// probe (<see cref="OtOpcUaNodeManager.NotifierDeliveryStall"/>). The probe is consulted only at a notifier that has
|
||||
/// event items (the hop the per-item delivery serves), so the stalled notifier must carry at least one.
|
||||
/// </remarks>
|
||||
internal sealed class NotifierStall : IDisposable
|
||||
{
|
||||
private readonly OtOpcUaNodeManager _nm;
|
||||
private readonly TaskCompletionSource _released = new(TaskCreationOptions.RunContinuationsAsynchronously);
|
||||
private readonly Func<NodeState, Task?> _probe;
|
||||
|
||||
private NotifierStall(OtOpcUaNodeManager nm, string notifierId)
|
||||
{
|
||||
_nm = nm;
|
||||
_probe = node => !_released.Task.IsCompleted && node.NodeId.IdentifierAsString == notifierId ? _released.Task : null;
|
||||
nm.NotifierDeliveryStall = _probe;
|
||||
}
|
||||
|
||||
/// <summary>Starts stalling delivery at the notifier whose NodeId identifier is <paramref name="notifierId"/>.</summary>
|
||||
public static NotifierStall Hold(OtOpcUaNodeManager nm, string notifierId) => new(nm, notifierId);
|
||||
|
||||
/// <summary>Lets every stalled walk continue (on pool threads, off <c>Lock</c>).</summary>
|
||||
public void Release() => _released.TrySetResult();
|
||||
|
||||
public void Dispose()
|
||||
{
|
||||
Release();
|
||||
if (ReferenceEquals(_nm.NotifierDeliveryStall, _probe)) _nm.NotifierDeliveryStall = null;
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user