Joseph Doherty
65ffa1dad4
The reconnect:false case sent its writes straight after ActorOf and gave each reply a fixed 1 s, so the window also covered the actor's start and first dispatch. A slow start (cold JIT, starved pool in a full solution run) missed it. The reconnect:true case was shielded by its 3 s wait for Connected. The test now waits until the connect attempt the writes race against has been issued (ConnectAsync entered, held open by the gate), so the actor is provably mid-attempt and cannot connect before the test releases it. Reply bounds are 10 s barriers, not part of the property: a write that was queued instead of rejected can only be answered after the gate opens, which never happens first. Evidence: with a 1.2 s slow actor start modelled (UpdateConnectionHealth Connecting sleeps), the old body passed 0/30 and the new one 30/30. Unmodified, the new test passed 40/40 under 14 busy-loop CPU load. Mutations still caught: Connecting dropping writes (timeout) and Connecting performing writes. Claude-Session: https://claude.ai/code/session_01GT67AEHd2Acv1YPwGfXEoV
385 lines
23 KiB
C#
385 lines
23 KiB
C#
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Services;
|
|
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 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>();
|
|
}
|
|
|
|
/// <summary>Verifies that a resolution tick does not duplicate an initial subscribe that is already in flight.</summary>
|
|
[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(); }
|
|
}
|
|
|
|
/// <summary>Verifies that an in-flight handle still serves a different instance after its original owner dies.</summary>
|
|
[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");
|
|
}
|
|
|
|
/// <summary>Verifies that a subscriber terminating during subscribe releases the late handle without resurrecting the subscription.</summary>
|
|
[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);
|
|
}
|
|
|
|
/// <summary>Verifies that a replacement during subscribe adopts only the tags that are still requested.</summary>
|
|
[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);
|
|
}
|
|
|
|
/// <summary>Verifies that a subscriber terminating during resolution releases the late handle.</summary>
|
|
[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);
|
|
}
|
|
|
|
/// <summary>Verifies that retried subscribes while disconnected keep only the latest request.</summary>
|
|
[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");
|
|
}
|
|
|
|
/// <summary>Verifies that writes while disconnected are rejected and never replayed after reconnect.</summary>
|
|
/// <param name="reconnect">Whether the connection is re-established after the rejected write.</param>
|
|
/// <remarks>
|
|
/// ScadaBridge#101: the writes used to be sent straight after <c>ActorOf</c> and their
|
|
/// replies awaited for a fixed 1 s, so in the <c>reconnect: false</c> case that window
|
|
/// also had to cover the actor's start and its first dispatch — a starved thread pool
|
|
/// in a full solution run could miss it. The test now waits on a barrier instead: the
|
|
/// connect attempt the writes race against has been ISSUED (the adapter's
|
|
/// <c>ConnectAsync</c> was entered) and is held open by <c>gate</c>, so the actor is
|
|
/// provably mid-attempt and cannot connect until the test releases it. The reply bound
|
|
/// is not part of the property: a write that was queued instead of rejected can only
|
|
/// be answered after the gate opens, which never happens before the bound expires.
|
|
/// </remarks>
|
|
[Theory]
|
|
[InlineData(false)]
|
|
[InlineData(true)]
|
|
public void DisconnectedWrites_AreRejectedAndNeverReplayed(bool reconnect)
|
|
{
|
|
var adapter = Adapter();
|
|
var gate = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
var attemptIssued = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
void HoldConnectAttempts() => adapter.ConnectAsync(Arg.Any<IDictionary<string, string>>(), Arg.Any<CancellationToken>())
|
|
.Returns(_ => { attemptIssued.TrySetResult(); return gate.Task; });
|
|
if (!reconnect) HoldConnectAttempts();
|
|
var actor = Actor(adapter);
|
|
if (reconnect)
|
|
{
|
|
AwaitCondition(() => HealthReported(ConnectionHealth.Connected), BarrierTimeout);
|
|
HoldConnectAttempts();
|
|
adapter.Disconnected += Raise.Event<Action>();
|
|
AwaitCondition(() => HealthReported(ConnectionHealth.Disconnected), BarrierTimeout);
|
|
}
|
|
AwaitCondition(() => attemptIssued.Task.IsCompleted, BarrierTimeout);
|
|
actor.Tell(new WriteTagRequest("single", "conn", "tag", 42, DateTimeOffset.UtcNow));
|
|
ExpectMsg<WriteTagResponse>(r => !r.Success && r.CorrelationId == "single", BarrierTimeout);
|
|
actor.Tell(new WriteTagBatchRequest("batch", "conn", new Dictionary<string, object?> { ["tag"] = 42 }, null, null, DateTimeOffset.UtcNow));
|
|
ExpectMsg<WriteTagBatchResponse>(r => !r.Success && r.CorrelationId == "batch", BarrierTimeout);
|
|
gate.SetResult();
|
|
actor.Tell(Subscribe());
|
|
ExpectMsg<SubscribeTagsResponse>(BarrierTimeout);
|
|
Assert.DoesNotContain(adapter.ReceivedCalls(), c => c.GetMethodInfo().Name is "WriteAsync" or "WriteBatchAsync");
|
|
}
|
|
|
|
// Upper bound for barrier waits; never a property under test (see ScadaBridge#101).
|
|
private static readonly TimeSpan BarrierTimeout = TimeSpan.FromSeconds(10);
|
|
|
|
private bool HealthReported(ConnectionHealth state) => _health.ReceivedCalls().Any(c =>
|
|
c.GetMethodInfo().Name == "UpdateConnectionHealth" && c.GetArguments()[1] is ConnectionHealth h && h == state);
|
|
|
|
/// <summary>Verifies that a subscriber terminating releases the handle of its last tag.</summary>
|
|
[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));
|
|
}
|
|
|
|
/// <summary>Verifies that an unsubscribe from the old owner does not remove the replacement's tags.</summary>
|
|
[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");
|
|
}
|
|
|
|
/// <summary>Verifies that a duplicate completion with a live handle does not unsubscribe it.</summary>
|
|
[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");
|
|
}
|
|
|
|
/// <summary>Verifies that a source-scoped snapshot end is not broadcast to other sources.</summary>
|
|
[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));
|
|
}
|
|
/// <summary>Verifies that a second subscriber receives the standing alarm snapshot.</summary>
|
|
/// <param name="protocol">The data connection protocol under test.</param>
|
|
[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);
|
|
}
|
|
|
|
/// <summary>Verifies that a subscriber joining mid-snapshot receives the prefix already seen.</summary>
|
|
/// <param name="scoped">Whether the snapshot is scoped to a single source.</param>
|
|
[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);
|
|
}
|
|
|
|
/// <summary>Verifies that replay rejects old-epoch rows and preserves a newer live clear.</summary>
|
|
/// <param name="staleEpoch">Whether the replayed rows belong to a stale epoch.</param>
|
|
[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);
|
|
}
|
|
|
|
}
|