Joseph Doherty
d9b23e0b95
Condition events reaching one of our notifiers no longer go through the SDK MonitoredNode2 channel: the walk serves the attached items itself, per item and per event (audit guard, uncached INodeManager3 verdict, item gate/cut, QueueEvent), as the Server-object path does. No notifier of ours defers any more; the #838 deferral queue and the #854 cut stay as the safety net and folder items share the cut (RefreshDirectItems, flag on and off). An event whose walk suspends is marked deferred before it suspends. Tests produce deferral with a notifier-stall probe instead of holding a folder item's SDK lock (which would now block the reporter). New FolderConditionRefreshWireTests; design note in docs/plans. Claude-Session: https://claude.ai/code/session_01GT67AD99zQQqpLCVm3p7xk
628 lines
33 KiB
C#
628 lines
33 KiB
C#
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>
|
|
/// 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)}"));
|
|
}
|
|
|
|
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,
|
|
};
|
|
}
|