Joseph Doherty
7783fadae1
- Both plain ConditionRefresh paths (Server object and folder) skip a condition a deploy removed or replaced between the off-Lock walk and the queueing (the SDK base did not either; only the cut path checked). - A failure delivering at one notifier hop is logged and the walk continues to the remaining hops; a failure deciding the Server-object items withholds only them. - Test: a condition removed during a plain refresh (folder + Server object). Claude-Session: https://claude.ai/code/session_01GT67AD99zQQqpLCVm3p7xk
842 lines
44 KiB
C#
842 lines
44 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>
|
|
/// #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,
|
|
};
|
|
}
|