Joseph Doherty
9a87d7a328
Second slice of the 2026-09-09 whole-system review remediation
(docs/plans/2026-09-09-arch-review.md §10), on top of the High slice 146a1323.
Runtime/devices: SR-2 alarm on-trigger integration scope; SR-3 correlated NativeAlarmActor
subscribe retry; SR-4 immutable audited DB target; SR-5 ordered InstanceActor stop/flush;
SR-12 rehydration fence; SR-13 S&F gate recheck per delivery; DC-4 no OPC UA security
downgrade; DC-5 alarm snapshot for every new subscriber; ORCH-1 sync-dispose joins the pump.
Management/UI: MG-2 site scope on audit tree + Flow Debugger Replay; MG-3 stale-probe memo
removed; MG-4 server-owned bulk-deploy deadline; UI-5 sandbox runs in a killable child
process (new CentralUI.SandboxWorker project); UI-14 bounded sandbox output/IPC.
Edge/audit: EI-1 relay audit ownership; EI-2 SkipBodyCapture suppresses error text; EI-3/EI-11
error-field redaction + caps; EI-4 single relay deadline, pre-execution-only retry; EI-5
authenticated relay provenance; EI-10 body cap before audit read.
Persistence/contracts: DL-1 global Instance.UniqueName unique index + conflict translation;
DL-2 DataConnection protocol config encrypted at rest (migration
20260909203046_EnforceGlobalInstanceNamesAndProtectProtocolConfiguration — coordinated
central upgrade, old binaries cannot read protected rows); DL-3 bounded Replay paging;
DL-6 cross-site Overwrite of a deployed instance rejected; XC-4 proto descriptor contract test.
Config/hygiene: XC-1 central listener isolation by local port; XC-2/XC-3 inert SMTP + cron
keys removed from rig settings; CM-2 env2 bootstrap guard enabled; DC-6 deploy site exit code;
DC-7 notifier retries only explicit standby rejection; XC-5 ClusterClient purged from docs.
Gates recorded in the report: build 0/0, 9 397 passed / 0 failed, rig redeployed on all
eight nodes with the migration applied. 46 Low findings remain open.
346 lines
20 KiB
C#
346 lines
20 KiB
C#
using Akka.Actor;
|
|
using Akka.TestKit.Xunit2;
|
|
using NSubstitute;
|
|
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Protocol;
|
|
using ZB.MOM.WW.ScadaBridge.Commons.Messages.DataConnection;
|
|
using ZB.MOM.WW.ScadaBridge.Commons.Types.Alarms;
|
|
using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums;
|
|
using ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Actors;
|
|
using ZB.MOM.WW.ScadaBridge.HealthMonitoring;
|
|
|
|
namespace ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests.Actors;
|
|
|
|
public class HighReviewRegressionTests : TestKit
|
|
{
|
|
private readonly ISiteHealthCollector _health = Substitute.For<ISiteHealthCollector>();
|
|
private static IDataConnection Adapter()
|
|
{
|
|
var adapter = Substitute.For<IDataConnection, IAlarmSubscribableConnection>();
|
|
adapter.ConnectAsync(Arg.Any<IDictionary<string, string>>(), Arg.Any<CancellationToken>()).Returns(Task.CompletedTask);
|
|
adapter.SubscribeAsync(Arg.Any<string>(), Arg.Any<SubscriptionCallback>(), Arg.Any<CancellationToken>()).Returns("handle");
|
|
adapter.ReadAsync(Arg.Any<string>(), Arg.Any<CancellationToken>()).Returns(new ReadResult(false, null, null));
|
|
return adapter;
|
|
}
|
|
private IActorRef Actor(IDataConnection adapter) => Sys.ActorOf(Props.Create(() => new DataConnectionActor(
|
|
"conn", adapter, new DataConnectionOptions { ReconnectInterval = TimeSpan.FromMilliseconds(50), TagResolutionRetryInterval = TimeSpan.FromHours(1), SeedReadMaxAttempts = 1 },
|
|
_health, Substitute.For<IDataConnectionFactory>(), "OpcUa")));
|
|
private void Connected() => AwaitCondition(() => _health.ReceivedCalls().Any(c => c.GetMethodInfo().Name == "UpdateConnectionHealth" && c.GetArguments()[1] is ConnectionHealth.Connected));
|
|
private static SubscribeTagsRequest Subscribe(string correlation = "s") => new(correlation, "instance", "conn", ["tag"], DateTimeOffset.UtcNow);
|
|
|
|
private DataConnectionHealthReport Health(IActorRef actor)
|
|
{
|
|
actor.Tell(new DataConnectionActor.GetHealthReport());
|
|
return ExpectMsg<DataConnectionHealthReport>();
|
|
}
|
|
|
|
[Fact]
|
|
public void ResolutionTick_DoesNotDuplicateAnInitialSubscribeAlreadyInFlight()
|
|
{
|
|
var adapter = new FakeBatchDataConnection();
|
|
adapter.FailingTags.Add("tag");
|
|
var actor = Actor(adapter);
|
|
var first = CreateTestProbe();
|
|
actor.Tell(Subscribe(), first.Ref);
|
|
first.ExpectMsg<TagValueUpdate>();
|
|
first.ExpectMsg<SubscribeTagsResponse>();
|
|
var gate = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
adapter.SubscribeGate = gate.Task;
|
|
adapter.FailingTags.Clear();
|
|
var second = CreateTestProbe();
|
|
actor.Tell(Subscribe("second") with { InstanceUniqueName = "second-instance" }, second.Ref);
|
|
AwaitCondition(() => adapter.SubscribeBatches.Count == 2);
|
|
var unexpected = CreateTestProbe();
|
|
adapter.BatchSubscribeThrows = () => { unexpected.Ref.Tell("duplicate subscribe"); return null; };
|
|
actor.Tell(new DataConnectionActor.RetryTagResolution());
|
|
try { unexpected.ExpectNoMsg(TimeSpan.FromMilliseconds(200)); }
|
|
finally { gate.SetResult(); }
|
|
}
|
|
|
|
[Fact]
|
|
public void OriginalOwnerDies_InFlightHandleStillServesDifferentInstance()
|
|
{
|
|
var adapter = Adapter();
|
|
var gate = new TaskCompletionSource<string>(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
adapter.SubscribeAsync("tag", Arg.Any<SubscriptionCallback>(), Arg.Any<CancellationToken>()).Returns(gate.Task);
|
|
var actor = Actor(adapter);
|
|
var a = CreateTestProbe(); var b = CreateTestProbe();
|
|
actor.Tell(Subscribe(), a.Ref);
|
|
AwaitCondition(() => adapter.ReceivedCalls().Any(c => c.GetMethodInfo().Name == "SubscribeAsync"));
|
|
actor.Tell(Subscribe("b") with { InstanceUniqueName = "other-instance" }, b.Ref);
|
|
b.ExpectMsg<SubscribeTagsResponse>();
|
|
Health(actor); // mailbox barrier after B's registration
|
|
var beforeDeath = _health.ReceivedCalls().Count(c => c.GetMethodInfo().Name == "UpdateTagResolution");
|
|
Sys.Stop(a.Ref);
|
|
AwaitCondition(() => _health.ReceivedCalls().Count(c => c.GetMethodInfo().Name == "UpdateTagResolution") > beforeDeath);
|
|
// The remaining instance still needs the tag, even though A started the I/O.
|
|
Assert.Equal(1, Health(actor).TotalSubscribedTags);
|
|
gate.SetResult("shared-handle");
|
|
AwaitAssert(() => Assert.Equal(1, Health(actor).ResolvedTags), TimeSpan.FromSeconds(1));
|
|
Assert.DoesNotContain(adapter.ReceivedCalls(), c => c.GetMethodInfo().Name == "UnsubscribeAsync");
|
|
}
|
|
|
|
[Fact]
|
|
public void SubscriberTermination_DuringSubscribe_ReleasesLateHandleWithoutResurrection()
|
|
{
|
|
var adapter = Adapter();
|
|
var gate = new TaskCompletionSource<string>(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
adapter.SubscribeAsync("tag", Arg.Any<SubscriptionCallback>(), Arg.Any<CancellationToken>()).Returns(gate.Task);
|
|
var actor = Actor(adapter);
|
|
var subscriber = CreateTestProbe();
|
|
actor.Tell(Subscribe(), subscriber.Ref);
|
|
AwaitCondition(() => adapter.ReceivedCalls().Any(c => c.GetMethodInfo().Name == "SubscribeAsync"));
|
|
Assert.Equal(1, Health(actor).TotalSubscribedTags);
|
|
Sys.Stop(subscriber.Ref);
|
|
AwaitAssert(() => Assert.Equal(0, Health(actor).TotalSubscribedTags));
|
|
gate.SetResult("late-handle");
|
|
AwaitCondition(() => adapter.ReceivedCalls().Any(c => c.GetMethodInfo().Name == "UnsubscribeAsync" && Equals(c.GetArguments()[0], "late-handle")));
|
|
Assert.Equal(0, Health(actor).ResolvedTags);
|
|
}
|
|
|
|
[Fact]
|
|
public void Replacement_DuringSubscribe_AdoptsOnlyStillRequestedTags()
|
|
{
|
|
var gate = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
var adapter = new FakeBatchDataConnection { SubscribeGate = gate.Task };
|
|
var actor = Actor(adapter);
|
|
var old = CreateTestProbe(); var current = CreateTestProbe();
|
|
actor.Tell(new SubscribeTagsRequest("old", "instance", "conn", ["keep", "drop"], DateTimeOffset.UtcNow), old.Ref);
|
|
AwaitCondition(() => adapter.SubscribeBatches.Count == 1);
|
|
actor.Tell(new SubscribeTagsRequest("new", "instance", "conn", ["keep"], DateTimeOffset.UtcNow), current.Ref);
|
|
current.ExpectMsg<SubscribeTagsResponse>();
|
|
Assert.Equal(1, Health(actor).TotalSubscribedTags);
|
|
gate.SetResult();
|
|
AwaitCondition(() => adapter.UnsubscribeBatches.SelectMany(b => b).Contains("sub-2"));
|
|
AwaitAssert(() => Assert.Equal(1, Health(actor).ResolvedTags));
|
|
Assert.DoesNotContain("sub-1", adapter.UnsubscribeBatches.SelectMany(b => b));
|
|
Sys.Stop(old.Ref);
|
|
Assert.Equal(1, Health(actor).TotalSubscribedTags);
|
|
}
|
|
|
|
[Fact]
|
|
public void SubscriberTermination_DuringResolution_ReleasesLateHandle()
|
|
{
|
|
var adapter = new FakeBatchDataConnection();
|
|
adapter.FailingTags.Add("tag");
|
|
var options = new DataConnectionOptions { TagResolutionRetryInterval = TimeSpan.FromMilliseconds(500), SeedReadMaxAttempts = 1 };
|
|
var actor = Sys.ActorOf(Props.Create(() => new DataConnectionActor("conn", adapter, options, _health, Substitute.For<IDataConnectionFactory>(), "OpcUa")));
|
|
var subscriber = CreateTestProbe();
|
|
actor.Tell(Subscribe(), subscriber.Ref);
|
|
subscriber.ExpectMsg<TagValueUpdate>();
|
|
subscriber.ExpectMsg<SubscribeTagsResponse>();
|
|
var gate = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
adapter.SubscribeGate = gate.Task;
|
|
adapter.FailingTags.Clear();
|
|
AwaitCondition(() => adapter.SubscribeBatches.Count == 2);
|
|
Sys.Stop(subscriber.Ref);
|
|
AwaitAssert(() => Assert.Equal(0, Health(actor).TotalSubscribedTags));
|
|
gate.SetResult();
|
|
AwaitCondition(() => adapter.UnsubscribeBatches.SelectMany(b => b).Any());
|
|
Assert.Equal(0, Health(actor).ResolvedTags);
|
|
}
|
|
|
|
[Fact]
|
|
public void DisconnectedSubscribeRetries_KeepOnlyLatestRequest()
|
|
{
|
|
var adapter = Adapter();
|
|
var gate = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
adapter.ConnectAsync(Arg.Any<IDictionary<string, string>>(), Arg.Any<CancellationToken>()).Returns(gate.Task);
|
|
var actor = Actor(adapter);
|
|
for (var i = 0; i < 100; i++) actor.Tell(Subscribe(i.ToString()));
|
|
Health(actor); // all retries have reached the disconnected-state queue
|
|
gate.SetResult();
|
|
ExpectMsg<SubscribeTagsResponse>(r => r.CorrelationId == "99");
|
|
ExpectNoMsg(TimeSpan.FromMilliseconds(200));
|
|
Assert.Single(adapter.ReceivedCalls(), c => c.GetMethodInfo().Name == "SubscribeAsync");
|
|
}
|
|
|
|
[Theory]
|
|
[InlineData(false)]
|
|
[InlineData(true)]
|
|
public void DisconnectedWrites_AreRejectedAndNeverReplayed(bool reconnect)
|
|
{
|
|
var adapter = Adapter();
|
|
var gate = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
if (!reconnect) adapter.ConnectAsync(Arg.Any<IDictionary<string, string>>(), Arg.Any<CancellationToken>()).Returns(gate.Task);
|
|
var actor = Actor(adapter);
|
|
if (reconnect)
|
|
{
|
|
Connected();
|
|
adapter.ConnectAsync(Arg.Any<IDictionary<string, string>>(), Arg.Any<CancellationToken>()).Returns(gate.Task);
|
|
adapter.Disconnected += Raise.Event<Action>();
|
|
AwaitCondition(() => _health.ReceivedCalls().Any(c => c.GetMethodInfo().Name == "UpdateConnectionHealth" && c.GetArguments()[1] is ConnectionHealth.Disconnected));
|
|
}
|
|
actor.Tell(new WriteTagRequest("single", "conn", "tag", 42, DateTimeOffset.UtcNow));
|
|
ExpectMsg<WriteTagResponse>(r => !r.Success && r.CorrelationId == "single", TimeSpan.FromSeconds(1));
|
|
actor.Tell(new WriteTagBatchRequest("batch", "conn", new Dictionary<string, object?> { ["tag"] = 42 }, null, null, DateTimeOffset.UtcNow));
|
|
ExpectMsg<WriteTagBatchResponse>(r => !r.Success && r.CorrelationId == "batch", TimeSpan.FromSeconds(1));
|
|
gate.SetResult();
|
|
actor.Tell(Subscribe());
|
|
ExpectMsg<SubscribeTagsResponse>();
|
|
Assert.DoesNotContain(adapter.ReceivedCalls(), c => c.GetMethodInfo().Name is "WriteAsync" or "WriteBatchAsync");
|
|
}
|
|
|
|
[Fact]
|
|
public void SubscriberTermination_ReleasesLastTagHandle()
|
|
{
|
|
var adapter = Adapter();
|
|
var actor = Actor(adapter);
|
|
var subscriber = CreateTestProbe();
|
|
actor.Tell(Subscribe(), subscriber.Ref);
|
|
subscriber.ExpectMsg<SubscribeTagsResponse>();
|
|
Sys.Stop(subscriber.Ref);
|
|
AwaitCondition(() => adapter.ReceivedCalls().Any(c => c.GetMethodInfo().Name == "UnsubscribeAsync" && Equals(c.GetArguments()[0], "handle")), TimeSpan.FromSeconds(1));
|
|
}
|
|
|
|
[Fact]
|
|
public void OldOwnerUnsubscribe_DoesNotRemoveReplacementTags()
|
|
{
|
|
var adapter = Adapter();
|
|
SubscriptionCallback? callback = null;
|
|
adapter.SubscribeAsync("tag", Arg.Do<SubscriptionCallback>(c => callback = c), Arg.Any<CancellationToken>()).Returns("handle");
|
|
var actor = Actor(adapter);
|
|
var old = CreateTestProbe();
|
|
var current = CreateTestProbe();
|
|
actor.Tell(Subscribe("old"), old.Ref);
|
|
old.ExpectMsg<SubscribeTagsResponse>();
|
|
actor.Tell(Subscribe("new"), current.Ref);
|
|
current.ExpectMsg<SubscribeTagsResponse>();
|
|
actor.Tell(new UnsubscribeTagsRequest("u", "instance", "conn", DateTimeOffset.UtcNow), old.Ref);
|
|
actor.Tell(Subscribe("barrier"), current.Ref);
|
|
current.ExpectMsg<SubscribeTagsResponse>();
|
|
callback!("tag", new TagValue(7, QualityCode.Good, DateTimeOffset.UtcNow));
|
|
current.ExpectMsg<TagValueUpdate>(u => Equals(u.Value, 7));
|
|
Assert.DoesNotContain(adapter.ReceivedCalls(), c => c.GetMethodInfo().Name == "UnsubscribeAsync");
|
|
}
|
|
|
|
[Fact]
|
|
public void DuplicateCompletion_WithLiveHandle_DoesNotUnsubscribeIt()
|
|
{
|
|
var adapter = Adapter();
|
|
var actor = Actor(adapter);
|
|
actor.Tell(Subscribe());
|
|
ExpectMsg<SubscribeTagsResponse>();
|
|
actor.Tell(new DataConnectionActor.SubscribeCompleted(Subscribe("duplicate"), TestActor,
|
|
[new DataConnectionActor.SubscribeTagResult("tag", false, true, "handle", null)], [], 0, adapter));
|
|
ExpectMsg<SubscribeTagsResponse>();
|
|
Assert.DoesNotContain(adapter.ReceivedCalls(), c => c.GetMethodInfo().Name == "UnsubscribeAsync");
|
|
}
|
|
|
|
[Fact]
|
|
public void SourceScopedSnapshotEnd_IsNotBroadcastToOtherSources()
|
|
{
|
|
var adapter = Adapter();
|
|
var alarm = (IAlarmSubscribableConnection)adapter;
|
|
AlarmTransitionCallback? callback = null;
|
|
alarm.SubscribeAlarmsAsync(Arg.Any<string>(), Arg.Any<string?>(), Arg.Do<AlarmTransitionCallback>(c => callback = c), Arg.Any<CancellationToken>()).Returns("alarm-handle");
|
|
var actor = Actor(adapter);
|
|
var a = CreateTestProbe(); var b = CreateTestProbe();
|
|
actor.Tell(new SubscribeAlarmsRequest("a", "a", "conn", "A", null, DateTimeOffset.UtcNow), a.Ref);
|
|
a.ExpectMsg<SubscribeAlarmsResponse>();
|
|
actor.Tell(new SubscribeAlarmsRequest("b", "b", "conn", "B", null, DateTimeOffset.UtcNow), b.Ref);
|
|
b.ExpectMsg<SubscribeAlarmsResponse>();
|
|
callback!(new NativeAlarmTransition("", "B", "", AlarmTransitionKind.SnapshotComplete,
|
|
new AlarmConditionState(false, false, null, AlarmShelveState.Unshelved, false, 0), "", "", "", "", "", null, DateTimeOffset.UtcNow, "", ""));
|
|
b.ExpectMsg<NativeAlarmTransitionUpdate>();
|
|
a.ExpectNoMsg(TimeSpan.FromMilliseconds(200));
|
|
}
|
|
[Theory]
|
|
[InlineData("OpcUa")]
|
|
[InlineData("MxGateway")]
|
|
public void DC5_SecondSubscriberReceivesStandingAlarmSnapshot(string protocol)
|
|
{
|
|
var adapter = Adapter();
|
|
AlarmTransitionCallback? callback = null;
|
|
((IAlarmSubscribableConnection)adapter).SubscribeAlarmsAsync(Arg.Any<string>(), Arg.Any<string?>(),
|
|
Arg.Do<AlarmTransitionCallback>(c => callback = c), Arg.Any<CancellationToken>()).Returns("alarm-handle");
|
|
var actor = Sys.ActorOf(Props.Create(() => new DataConnectionActor("conn", adapter, new DataConnectionOptions(),
|
|
_health, Substitute.For<IDataConnectionFactory>(), protocol)));
|
|
var a = CreateTestProbe(); var b = CreateTestProbe();
|
|
actor.Tell(new SubscribeAlarmsRequest("a", "a", "conn", "A", null, DateTimeOffset.UtcNow), a.Ref);
|
|
a.ExpectMsg<SubscribeAlarmsResponse>();
|
|
var alarm = new NativeAlarmTransition("A.Hi", "A", "Limit", AlarmTransitionKind.Snapshot,
|
|
new AlarmConditionState(true, false, null, AlarmShelveState.Unshelved, false, 800), "", "", "", "", "", null, DateTimeOffset.UtcNow, "", "");
|
|
if (protocol == "OpcUa")
|
|
{
|
|
alarm = alarm with { SnapshotSourceReference = "A", SnapshotId = Guid.NewGuid() };
|
|
callback!(alarm with { SourceReference = "", Kind = AlarmTransitionKind.SnapshotStart });
|
|
a.ExpectMsg<NativeAlarmTransitionUpdate>();
|
|
}
|
|
callback!(alarm);
|
|
callback!(alarm with { SourceReference = "", Kind = AlarmTransitionKind.SnapshotComplete });
|
|
a.ExpectMsg<NativeAlarmTransitionUpdate>(); a.ExpectMsg<NativeAlarmTransitionUpdate>();
|
|
if (protocol == "OpcUa")
|
|
{
|
|
// A duplicate end must not erase the DCL replay cache either.
|
|
callback!(alarm with { SourceReference = "", Kind = AlarmTransitionKind.SnapshotComplete });
|
|
a.ExpectMsg<NativeAlarmTransitionUpdate>();
|
|
}
|
|
actor.Tell(new SubscribeAlarmsRequest("b", "b", "conn", "A", null, DateTimeOffset.UtcNow), b.Ref);
|
|
var received = b.ReceiveN(4, TimeSpan.FromSeconds(2));
|
|
Assert.Contains(received.OfType<NativeAlarmTransitionUpdate>(), m => m.Transition.Kind == AlarmTransitionKind.Snapshot && m.Transition.SourceReference == "A.Hi");
|
|
Assert.Contains(received.OfType<NativeAlarmTransitionUpdate>(), m => m.Transition.Kind == AlarmTransitionKind.SnapshotComplete);
|
|
}
|
|
|
|
[Theory]
|
|
[InlineData(false)]
|
|
[InlineData(true)]
|
|
public void DC5_MidSnapshotSubscriberReceivesAlreadySeenPrefix(bool scoped)
|
|
{
|
|
var adapter = Adapter();
|
|
AlarmTransitionCallback? callback = null;
|
|
((IAlarmSubscribableConnection)adapter).SubscribeAlarmsAsync(Arg.Any<string>(), Arg.Any<string?>(),
|
|
Arg.Do<AlarmTransitionCallback>(c => callback = c), Arg.Any<CancellationToken>()).Returns("alarm-handle");
|
|
var actor = Actor(adapter);
|
|
var a = CreateTestProbe(); var b = CreateTestProbe();
|
|
actor.Tell(new SubscribeAlarmsRequest("a", "a", "conn", "A", null, DateTimeOffset.UtcNow), a.Ref);
|
|
a.ExpectMsg<SubscribeAlarmsResponse>();
|
|
var alarm = new NativeAlarmTransition("A.Hi", "A", "Limit", AlarmTransitionKind.Snapshot,
|
|
new AlarmConditionState(true, false, null, AlarmShelveState.Unshelved, false, 800), "", "", "", "", "", null, DateTimeOffset.UtcNow, "", "",
|
|
SnapshotSourceReference: scoped ? "A" : null, SnapshotId: scoped ? Guid.NewGuid() : null);
|
|
if (scoped)
|
|
{
|
|
callback!(alarm with { SourceReference = "", Kind = AlarmTransitionKind.SnapshotStart });
|
|
a.ExpectMsg<NativeAlarmTransitionUpdate>();
|
|
}
|
|
callback!(alarm);
|
|
a.ExpectMsg<NativeAlarmTransitionUpdate>();
|
|
actor.Tell(new SubscribeAlarmsRequest("b", "b", "conn", "A", null, DateTimeOffset.UtcNow), b.Ref);
|
|
var prefix = b.ReceiveN(scoped ? 3 : 2, TimeSpan.FromSeconds(2));
|
|
Assert.Contains(prefix.OfType<NativeAlarmTransitionUpdate>(), m => m.Transition.SourceReference == "A.Hi");
|
|
callback!(alarm with { SourceReference = "", Kind = AlarmTransitionKind.SnapshotComplete });
|
|
var end = b.ExpectMsg<NativeAlarmTransitionUpdate>();
|
|
Assert.Equal(AlarmTransitionKind.SnapshotComplete, end.Transition.Kind);
|
|
Assert.Equal(alarm.SnapshotId, end.Transition.SnapshotId);
|
|
}
|
|
|
|
[Theory]
|
|
[InlineData(false)]
|
|
[InlineData(true)]
|
|
public void DC5_ReplayRejectsOldEpochRowsAndPreservesNewerLiveClear(bool staleEpoch)
|
|
{
|
|
var adapter = Adapter();
|
|
AlarmTransitionCallback? callback = null;
|
|
((IAlarmSubscribableConnection)adapter).SubscribeAlarmsAsync(Arg.Any<string>(), Arg.Any<string?>(),
|
|
Arg.Do<AlarmTransitionCallback>(c => callback = c), Arg.Any<CancellationToken>()).Returns("alarm-handle");
|
|
var actor = Actor(adapter);
|
|
var a = CreateTestProbe(); var b = CreateTestProbe();
|
|
actor.Tell(new SubscribeAlarmsRequest("a", "a", "conn", "A", null, DateTimeOffset.UtcNow), a.Ref);
|
|
a.ExpectMsg<SubscribeAlarmsResponse>();
|
|
var alarm = new NativeAlarmTransition("A.Hi", "A", "Limit", AlarmTransitionKind.Snapshot,
|
|
new AlarmConditionState(true, false, null, AlarmShelveState.Unshelved, false, 800), "", "", "", "", "", null, DateTimeOffset.UtcNow, "", "",
|
|
SnapshotSourceReference: "A", SnapshotId: Guid.NewGuid());
|
|
callback!(alarm with { SourceReference = "", Kind = AlarmTransitionKind.SnapshotStart });
|
|
if (staleEpoch) callback!(alarm with { SourceReference = "A.Old", SnapshotId = Guid.NewGuid() });
|
|
callback!(alarm);
|
|
if (!staleEpoch) callback!(alarm with { Kind = AlarmTransitionKind.Clear, Condition = alarm.Condition with { Active = false, Acknowledged = true }, TransitionTime = alarm.TransitionTime.AddSeconds(1) });
|
|
callback!(alarm with { SourceReference = "", Kind = AlarmTransitionKind.SnapshotComplete });
|
|
a.ReceiveN(4, TimeSpan.FromSeconds(2));
|
|
actor.Tell(new SubscribeAlarmsRequest("b", "b", "conn", "A", null, DateTimeOffset.UtcNow), b.Ref);
|
|
var replay = b.ReceiveN(staleEpoch ? 4 : 3, TimeSpan.FromSeconds(2));
|
|
var rows = replay.OfType<NativeAlarmTransitionUpdate>().Where(m => m.Transition.Kind == AlarmTransitionKind.Snapshot).ToList();
|
|
if (staleEpoch) Assert.Equal("A.Hi", Assert.Single(rows).Transition.SourceReference);
|
|
else Assert.Empty(rows);
|
|
}
|
|
|
|
}
|