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.Enums;
using ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Actors;
using ZB.MOM.WW.ScadaBridge.HealthMonitoring;
namespace ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests.Actors;
///
/// Arch-review 2026-08-19 P2-6 — the AdapterDisconnected gap in the
/// Connecting/Reconnecting states, and the generation fence that keeps a swapped-out adapter's
/// disconnect from disturbing its replacement.
///
///
/// The gap these pin. Only the Connected state handled AdapterDisconnected; in
/// Connecting and Reconnecting it fell through to Unhandled. A connect that succeeds and
/// drops again before its piped ConnectResult is applied therefore delivered the disconnect
/// FIRST, where it was discarded, and the ConnectResult(true) that followed declared the
/// connection up on an adapter that was already dead. Nothing recovers from there: an adapter
/// raises Disconnected at most once per successful connect (OpcUaDataConnection's
/// once-only _disconnectFired guard, reset only by the next successful ConnectAsync),
/// so on a connection with no subscriptions — nothing to fail at connection level and drive the
/// reconnect machine — the actor sat Connected on a dead adapter for the life of the process.
///
///
///
/// The fix must not weaken the Medium-wave generation fencing: a disconnect raised by an adapter
/// that has since been swapped out (failover, or the row-97 in-place reconfigure) must still be
/// dropped, or the endpoint the operator has just corrected AWAY from could knock the healthy
/// replacement back into a reconnect cycle.
///
///
///
/// Determinism note. The race is reproduced by ORDERING, not by timing: the adapter's
/// connect is held on a gate, the Disconnected event is raised while the actor sits idle in
/// Connecting, and a GetHealthReport round trip is used as a mailbox barrier proving the
/// disconnect has been processed before the connect result is released. No sleeps.
///
///
[Collection("DataConnectionManagerActor")]
public class DataConnectionDisconnectRaceTests : TestKit
{
private const string FirstEndpoint = "opc.tcp://first-plc:4840";
private const string SecondEndpoint = "opc.tcp://second-plc:4840";
private readonly ISiteHealthCollector _health = Substitute.For();
private readonly IDataConnectionFactory _factory = Substitute.For();
private readonly DataConnectionOptions _options = new()
{
ReconnectInterval = TimeSpan.FromMilliseconds(100),
TagResolutionRetryInterval = TimeSpan.FromMilliseconds(200),
WriteTimeout = TimeSpan.FromSeconds(5),
SeedReadMaxAttempts = 1,
SeedReadRetryDelay = TimeSpan.FromMilliseconds(10)
};
private static Dictionary Config(string endpoint) =>
new() { ["Endpoint"] = endpoint };
private IActorRef CreateConnectionActor(string name, IDataConnection adapter) =>
Sys.ActorOf(Props.Create(() =>
new DataConnectionActor(
name, adapter, _options, _health, _factory, "OpcUa",
Config(FirstEndpoint), null, 3)), name);
private static IDataConnection Adapter(Task connect)
{
var adapter = Substitute.For();
adapter.ConnectAsync(Arg.Any>(), Arg.Any())
.Returns(connect);
adapter.Status.Returns(ConnectionHealth.Connected);
adapter.SubscribeAsync(Arg.Any(), Arg.Any(), Arg.Any())
.Returns("sub-1");
adapter.ReadAsync(Arg.Any(), Arg.Any())
.Returns(new ReadResult(false, null, null));
return adapter;
}
private int HealthTransitions(string connectionName, ConnectionHealth health) =>
_health.ReceivedCalls().Count(c =>
c.GetMethodInfo().Name == nameof(ISiteHealthCollector.UpdateConnectionHealth) &&
c.GetArguments() is [string name, ConnectionHealth reported, ..] &&
name == connectionName && reported == health);
///
/// Mailbox barrier: the reply to a message sent AFTER was told
/// something proves the earlier message has already been processed (Akka mailboxes are FIFO
/// per sender pair, and every state of this actor answers GetHealthReport). Used
/// instead of a settle delay so the ordering the defect depends on is asserted, not waited on.
///
private void Barrier(IActorRef actor)
{
var probe = CreateTestProbe();
probe.Send(actor, new DataConnectionActor.GetHealthReport());
probe.ExpectMsg(TimeSpan.FromSeconds(5));
}
///
/// The headline case: the adapter connects and drops again before the actor has applied the
/// connect result. The actor must NOT settle in Connected — the adapter will never raise
/// Disconnected again, so Connected would be terminal — and must retry instead.
///
/// Fails before the fix: the disconnect was Unhandled in Connecting, the actor
/// went straight to Connected and never reconnected — exactly one ConnectAsync for the
/// life of the actor.
///
[Fact]
public void DisconnectBeforeConnectResult_RetriesInsteadOfGoingConnected()
{
const string name = "disconnect-before-connect-result";
var firstConnect = new TaskCompletionSource();
// Attempt #2 onward never completes, so "the actor did not go Connected on the dead
// session" is a stable assertion rather than a race against the next retry tick.
var laterConnects = new TaskCompletionSource();
var connects = 0;
var adapter = Substitute.For();
adapter.ConnectAsync(Arg.Any>(), Arg.Any())
.Returns(_ => Interlocked.Increment(ref connects) == 1 ? firstConnect.Task : laterConnects.Task);
adapter.Status.Returns(ConnectionHealth.Connected);
adapter.ReadAsync(Arg.Any(), Arg.Any())
.Returns(new ReadResult(false, null, null));
var actor = CreateConnectionActor(name, adapter);
// The actor is in Connecting with ConnectAsync in flight on the gate.
AwaitCondition(() => Volatile.Read(ref connects) >= 1,
TimeSpan.FromSeconds(5), TimeSpan.FromMilliseconds(10));
// The link drops. This is the real ordering: the adapter's keep-alive thread raises
// Disconnected while the connect's piped result is still on its way.
adapter.Disconnected += Raise.Event();
Barrier(actor);
// Only now does the (already obsolete) success land.
firstConnect.SetResult();
// The retry cycle runs, rather than the actor parking Connected on a dead adapter.
AwaitCondition(() => Volatile.Read(ref connects) >= 2,
TimeSpan.FromSeconds(5), TimeSpan.FromMilliseconds(10));
Assert.Equal(0, HealthTransitions(name, ConnectionHealth.Connected));
}
///
/// The Reconnecting twin of the case above — the same gap existed there, and a reconnect is
/// where a flapping endpoint spends most of its time.
///
/// Fails before the fix: the disconnect was Unhandled in Reconnecting and the
/// stale success re-subscribed and declared the connection up on the dead adapter.
///
[Fact]
public void DisconnectBeforeReconnectResult_StaysReconnecting()
{
const string name = "disconnect-before-reconnect-result";
var firstConnect = new TaskCompletionSource();
var secondConnect = new TaskCompletionSource();
// Attempt #3 onward never completes, so the actor parks in Reconnecting and the
// "did not go Connected on the dead session" assertion is stable rather than a race
// against the next retry tick.
var laterConnects = new TaskCompletionSource();
var connects = 0;
var adapter = Substitute.For();
adapter.ConnectAsync(Arg.Any>(), Arg.Any())
.Returns(_ => Interlocked.Increment(ref connects) switch
{
1 => firstConnect.Task,
2 => secondConnect.Task,
_ => laterConnects.Task
});
adapter.Status.Returns(ConnectionHealth.Connected);
adapter.ReadAsync(Arg.Any(), Arg.Any())
.Returns(new ReadResult(false, null, null));
var actor = CreateConnectionActor(name, adapter);
firstConnect.SetResult();
AwaitCondition(() => HealthTransitions(name, ConnectionHealth.Connected) >= 1,
TimeSpan.FromSeconds(5), TimeSpan.FromMilliseconds(10));
// Drop while Connected → Reconnecting, and the retry timer launches attempt #2.
adapter.Disconnected += Raise.Event();
AwaitCondition(() => Volatile.Read(ref connects) >= 2,
TimeSpan.FromSeconds(5), TimeSpan.FromMilliseconds(10));
// The reconnect drops again before its result is applied.
adapter.Disconnected += Raise.Event();
Barrier(actor);
secondConnect.SetResult();
// The demoted success re-arms the retry timer instead of declaring the link up again.
AwaitCondition(() => Volatile.Read(ref connects) >= 3,
TimeSpan.FromSeconds(5), TimeSpan.FromMilliseconds(10));
Assert.Equal(1, HealthTransitions(name, ConnectionHealth.Connected));
}
///
/// The fencing guard rail: a disconnect stamped with a superseded adapter generation must
/// still be dropped. SwapAdapter unhooks the outgoing adapter, but the event can
/// already be running on a background thread when it does, so the message can outlive the
/// swap — and acting on it would take a healthy, freshly reconfigured connection down.
///
///
/// Proof of failure. This one cannot be shown against the pre-fix tree by stashing:
/// AdapterDisconnected carried no generation at all before this change, so the test
/// would not compile. It was proved by a targeted revert instead — restoring
/// case AdapterDisconnected: HandleDisconnect(); break; in the Connected state while
/// keeping the stamped record — which makes the assertion below fail with 2 Disconnected
/// transitions instead of 1.
///
///
[Fact]
public void StaleGenerationDisconnect_IsIgnored()
{
const string name = "stale-generation-disconnect";
var first = Adapter(Task.CompletedTask);
var second = Adapter(Task.CompletedTask);
_factory.Create("OpcUa", Arg.Is>(d => d["Endpoint"] == SecondEndpoint))
.Returns(second);
var actor = CreateConnectionActor(name, first);
AwaitCondition(() => HealthTransitions(name, ConnectionHealth.Connected) >= 1,
TimeSpan.FromSeconds(5), TimeSpan.FromMilliseconds(10));
// In-place reconfigure: swaps the adapter and bumps the generation 0 → 1. The single
// Disconnected transition it publishes on the way through Reconnecting is the baseline.
actor.Tell(new ReconfigureConnectionCommand(name, "OpcUa", Config(SecondEndpoint), null, 3));
AwaitCondition(() => HealthTransitions(name, ConnectionHealth.Connected) >= 2,
TimeSpan.FromSeconds(5), TimeSpan.FromMilliseconds(10));
Assert.Equal(1, HealthTransitions(name, ConnectionHealth.Disconnected));
// The endpoint we have just moved away from finally reports its loss.
actor.Tell(new DataConnectionActor.AdapterDisconnected(0));
Barrier(actor);
Assert.Equal(1, HealthTransitions(name, ConnectionHealth.Disconnected));
Assert.Equal(2, HealthTransitions(name, ConnectionHealth.Connected));
}
}