test(dcl): barrier instead of a 1 s window in DisconnectedWrites_AreRejectedAndNeverReplayed (ScadaBridge#101)
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
This commit is contained in:
+28
-8
@@ -25,7 +25,6 @@ public class HighReviewRegressionTests : TestKit
|
|||||||
private IActorRef Actor(IDataConnection adapter) => Sys.ActorOf(Props.Create(() => new DataConnectionActor(
|
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 },
|
"conn", adapter, new DataConnectionOptions { ReconnectInterval = TimeSpan.FromMilliseconds(50), TagResolutionRetryInterval = TimeSpan.FromHours(1), SeedReadMaxAttempts = 1 },
|
||||||
_health, Substitute.For<IDataConnectionFactory>(), "OpcUa")));
|
_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 static SubscribeTagsRequest Subscribe(string correlation = "s") => new(correlation, "instance", "conn", ["tag"], DateTimeOffset.UtcNow);
|
||||||
|
|
||||||
private DataConnectionHealthReport Health(IActorRef actor)
|
private DataConnectionHealthReport Health(IActorRef actor)
|
||||||
@@ -163,6 +162,17 @@ public class HighReviewRegressionTests : TestKit
|
|||||||
|
|
||||||
/// <summary>Verifies that writes while disconnected are rejected and never replayed after reconnect.</summary>
|
/// <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>
|
/// <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]
|
[Theory]
|
||||||
[InlineData(false)]
|
[InlineData(false)]
|
||||||
[InlineData(true)]
|
[InlineData(true)]
|
||||||
@@ -170,25 +180,35 @@ public class HighReviewRegressionTests : TestKit
|
|||||||
{
|
{
|
||||||
var adapter = Adapter();
|
var adapter = Adapter();
|
||||||
var gate = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
|
var gate = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
|
||||||
if (!reconnect) adapter.ConnectAsync(Arg.Any<IDictionary<string, string>>(), Arg.Any<CancellationToken>()).Returns(gate.Task);
|
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);
|
var actor = Actor(adapter);
|
||||||
if (reconnect)
|
if (reconnect)
|
||||||
{
|
{
|
||||||
Connected();
|
AwaitCondition(() => HealthReported(ConnectionHealth.Connected), BarrierTimeout);
|
||||||
adapter.ConnectAsync(Arg.Any<IDictionary<string, string>>(), Arg.Any<CancellationToken>()).Returns(gate.Task);
|
HoldConnectAttempts();
|
||||||
adapter.Disconnected += Raise.Event<Action>();
|
adapter.Disconnected += Raise.Event<Action>();
|
||||||
AwaitCondition(() => _health.ReceivedCalls().Any(c => c.GetMethodInfo().Name == "UpdateConnectionHealth" && c.GetArguments()[1] is ConnectionHealth.Disconnected));
|
AwaitCondition(() => HealthReported(ConnectionHealth.Disconnected), BarrierTimeout);
|
||||||
}
|
}
|
||||||
|
AwaitCondition(() => attemptIssued.Task.IsCompleted, BarrierTimeout);
|
||||||
actor.Tell(new WriteTagRequest("single", "conn", "tag", 42, DateTimeOffset.UtcNow));
|
actor.Tell(new WriteTagRequest("single", "conn", "tag", 42, DateTimeOffset.UtcNow));
|
||||||
ExpectMsg<WriteTagResponse>(r => !r.Success && r.CorrelationId == "single", TimeSpan.FromSeconds(1));
|
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));
|
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));
|
ExpectMsg<WriteTagBatchResponse>(r => !r.Success && r.CorrelationId == "batch", BarrierTimeout);
|
||||||
gate.SetResult();
|
gate.SetResult();
|
||||||
actor.Tell(Subscribe());
|
actor.Tell(Subscribe());
|
||||||
ExpectMsg<SubscribeTagsResponse>();
|
ExpectMsg<SubscribeTagsResponse>(BarrierTimeout);
|
||||||
Assert.DoesNotContain(adapter.ReceivedCalls(), c => c.GetMethodInfo().Name is "WriteAsync" or "WriteBatchAsync");
|
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>
|
/// <summary>Verifies that a subscriber terminating releases the handle of its last tag.</summary>
|
||||||
[Fact]
|
[Fact]
|
||||||
public void SubscriberTermination_ReleasesLastTagHandle()
|
public void SubscriberTermination_ReleasesLastTagHandle()
|
||||||
|
|||||||
Reference in New Issue
Block a user