fix(opcua): condition events never block under the node-manager lock; folder-event Part 3 note (#838)
#838(a): each notifier's SDK MonitoredNode2 drains a bounded channel (4096, FullMode=Wait) and the synchronous NodeState.ReportEvent blocks when it is full. Condition events are reported under Lock, and the delivery task takes Lock on a verdict-cache miss (the SDK's own GetNodeMetadata, flag off too, and #826's IsNotifierEventVisibleTo). A burst of >4096 events under one Lock hold therefore deadlocked the node manager -- FolderEventFloodWireTests stalled at exactly 4097 events with enforcement on AND off (so it predates #826). ReportConditionEvent now goes through ReportEventWithoutBlocking: ReportEventAsync completes inline while channels have room (unchanged normal path); on overflow the report continues asynchronously holding no thread or lock, and later reports queue behind it so events keep report order. The deferred work holds only frozen snapshots and drains once Lock is released; client delivery stays bounded by each item's queue. #838(b): documented in docs/security.md that Part 3 ReceiveEvents applies to the event's EventType and SourceNode (the SDK check, kept first), not to the notifier; our decision is layered on top and non-condition events on a folder item fail closed under enforcement. The reflection item stays open.
This commit is contained in:
Joseph Doherty
+136
@@ -0,0 +1,136 @@
|
||||
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;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests;
|
||||
|
||||
/// <summary>
|
||||
/// Gitea #838 (follow-up to #826): the per-folder event ACL runs on the SDK's <c>MonitoredNode2</c> delivery task,
|
||||
/// and needs the node manager's <c>Lock</c> on a verdict-cache miss. That delivery task drains a BOUNDED channel
|
||||
/// (4096, <c>FullMode = Wait</c>), and the synchronous <c>NodeState.ReportEvent</c> the node manager calls under
|
||||
/// <c>Lock</c> BLOCKS when the channel is full. So a burst of more than 4096 events reported under one hold of
|
||||
/// <c>Lock</c>, starting at a verdict-cache miss, parks the consumer on <c>Lock</c> while the producer waits on the
|
||||
/// 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>
|
||||
public sealed class FolderEventFloodWireTests
|
||||
{
|
||||
private const string EqA = ServerEventRig.EqA;
|
||||
private const int FloodSize = 6000; // > the SDK's 4096-slot per-node channel
|
||||
|
||||
/// <param name="enforced">
|
||||
/// Per-folder ACL on or off. The hazard does NOT need #826: with enforcement off the SDK's own
|
||||
/// <c>ValidateEventRolePermissions</c> still reaches <c>GetNodeMetadata</c>, which takes <c>Lock</c>, on the
|
||||
/// first event to a new item.
|
||||
/// </param>
|
||||
[Theory]
|
||||
[InlineData(true)]
|
||||
[InlineData(false)]
|
||||
public async Task A_flood_beyond_the_channel_under_the_node_manager_lock_neither_deadlocks_nor_grows_unbounded(bool enforced)
|
||||
{
|
||||
var ct = TestContext.Current.CancellationToken;
|
||||
var rig = await ServerEventRig.StartAsync("Folder838Flood", encrypted: false, ct);
|
||||
var deadlocked = false;
|
||||
ISession? keyed = null;
|
||||
var reported = 0;
|
||||
try
|
||||
{
|
||||
rig.KeyRoles = [];
|
||||
rig.MaterialiseNative();
|
||||
NodeAcl[] grants = [KeyEquipmentGrant(EqA)];
|
||||
rig.Enforce(enforced, [ServerEventRig.Key()], grants);
|
||||
|
||||
keyed = await rig.ConnectAsync(anonymous: false, ct);
|
||||
var events = new RecordedEvents();
|
||||
var subscription = await RecordedEvents.SubscribeAsync(keyed, events, ct);
|
||||
var onA = RecordedEvents.AddEventItem(subscription, ServerEventRig.UnsNodeId(keyed, EqA));
|
||||
await subscription.ApplyChangesAsync(ct);
|
||||
onA.Status.Error.ShouldBeNull();
|
||||
|
||||
// No publishing: the server-side item queue (QueueSize 1000) is the only sink, as with a stalled client.
|
||||
await subscription.SetPublishingModeAsync(false, ct);
|
||||
|
||||
// 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 flood = Task.Factory.StartNew(() =>
|
||||
{
|
||||
lock (rig.Nm.Lock)
|
||||
{
|
||||
for (var i = 0; i < FloodSize; i++)
|
||||
{
|
||||
rig.Project(native: true, State($"f{i}"));
|
||||
Volatile.Write(ref reported, i + 1);
|
||||
}
|
||||
}
|
||||
}, ct, TaskCreationOptions.LongRunning, TaskScheduler.Default);
|
||||
|
||||
try
|
||||
{
|
||||
await flood.WaitAsync(TimeSpan.FromSeconds(60), ct);
|
||||
}
|
||||
catch (TimeoutException)
|
||||
{
|
||||
deadlocked = true;
|
||||
throw new ShouldAssertException(
|
||||
$"reporting {FloodSize} folder events under the node manager's Lock stalled after "
|
||||
+ $"{Volatile.Read(ref reported)} — the SDK's per-node event channel filled while its consumer waited "
|
||||
+ "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);
|
||||
|
||||
// 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.
|
||||
await subscription.SetPublishingModeAsync(true, ct);
|
||||
var last = $"f{FloodSize - 1}";
|
||||
await RecordedEvents.WaitUntilAsync(() => events.Has(onA, last));
|
||||
await Task.Delay(TimeSpan.FromMilliseconds(500), ct);
|
||||
var flooded = events.Of(onA)
|
||||
.Select(RecordedEvents.Message)
|
||||
.Where(m => m?.StartsWith('f') == true)
|
||||
.Select(m => int.Parse(m![1..], System.Globalization.CultureInfo.InvariantCulture))
|
||||
.ToList();
|
||||
flooded.Count.ShouldBeGreaterThan(0);
|
||||
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);
|
||||
|
||||
// And the node manager is live: a later event is decided by the per-folder ACL and delivered.
|
||||
rig.Project(native: true, State("after"));
|
||||
await RecordedEvents.WaitUntilAsync(() => events.Has(onA, "after"));
|
||||
}
|
||||
finally
|
||||
{
|
||||
// A deadlocked rig cannot be disposed (its Lock is held by a blocked thread) and neither can a session on it
|
||||
// (CloseSession needs the server); leak both rather than hang the run.
|
||||
if (!deadlocked)
|
||||
{
|
||||
keyed?.Dispose();
|
||||
await rig.DisposeAsync();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private static AlarmConditionSnapshot State(string message) =>
|
||||
new(true, false, false, 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,
|
||||
};
|
||||
}
|
||||
Reference in New Issue
Block a user