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)); } }