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;
///
/// Deferred register row 97 — in-place reconfiguration of a LIVE data connection.
///
///
/// The gap these pin. DataConnectionManagerActor.HandleCreateConnection is
/// create-only: it warns and returns for a name it already holds. So the Deployment Manager's
/// documented "a changed configuration re-issues the create so the DCL adopts it" was never true
/// on ANY node — a corrected endpoint or credential took effect only when the connection was next
/// built from scratch (node restart, singleton failover, or a remove-then-add), while the DM's
/// hash cache recorded it as applied. Discovered while validating rows 89/90 and recorded as
/// residual 1 of docs/plans/2026-08-17-dcl-connection-arbitration-decision.md.
///
///
///
/// Why the fix is a swap and not a restart. Stopping and re-creating the connection actor
/// is destructive: the subscription registry lives inside DataConnectionActor, and
/// InstanceActor cancels its per-connection subscribe-retry timer on the first success —
/// so nothing would ever re-subscribe. That trades "stale endpoint, data flowing" for "correct
/// endpoint, no data until redeploy". instead reuses
/// the primary→backup failover machinery: dispose the adapter, build one from the new config,
/// bump the adapter generation, and let the reconnect cycle re-issue every subscription from the
/// registry that was never touched.
///
///
[Collection("DataConnectionManagerActor")]
public class DataConnectionReconfigureTests : TestKit
{
private const string OldEndpoint = "opc.tcp://old-plc:4840";
private const string NewEndpoint = "opc.tcp://new-plc:4840";
private readonly DataConnectionOptions _options;
private readonly ISiteHealthCollector _healthCollector;
private readonly IDataConnectionFactory _factory;
public DataConnectionReconfigureTests()
: base(@"akka.loglevel = DEBUG")
{
_healthCollector = Substitute.For();
_factory = Substitute.For();
_options = new DataConnectionOptions
{
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, IDictionary primaryConfig) =>
Sys.ActorOf(Props.Create(() =>
new DataConnectionActor(
name, adapter, _options, _healthCollector, _factory, "OpcUa",
primaryConfig, null, 3)), name);
/// An adapter that connects, subscribes and reads without incident.
private static IDataConnection HealthyAdapter(string subscriptionId)
{
var adapter = Substitute.For();
adapter.ConnectAsync(Arg.Any>(), Arg.Any())
.Returns(Task.CompletedTask);
adapter.Status.Returns(ConnectionHealth.Connected);
adapter.SubscribeAsync(Arg.Any(), Arg.Any(), Arg.Any())
.Returns(subscriptionId);
adapter.ReadAsync(Arg.Any(), Arg.Any())
.Returns(new ReadResult(false, null, null));
return adapter;
}
private static void RaiseDisconnected(IDataConnection adapter) =>
adapter.Disconnected += Raise.Event();
private bool HasEnteredConnectedState(string connectionName) =>
_healthCollector.ReceivedCalls().Any(c =>
c.GetMethodInfo().Name == nameof(ISiteHealthCollector.UpdateConnectionHealth) &&
c.GetArguments() is [string name, ConnectionHealth.Connected, ..] &&
name == connectionName);
private bool CreatedAdapterFor(string endpoint) =>
_factory.ReceivedCalls().Any(c =>
c.GetMethodInfo().Name == nameof(IDataConnectionFactory.Create) &&
c.GetArguments()[1] is IDictionary d &&
d.TryGetValue("Endpoint", out var e) && e == endpoint);
private static bool Subscribed(IDataConnection adapter) =>
adapter.ReceivedCalls().Any(c => c.GetMethodInfo().Name == "SubscribeAsync");
private static bool Connected(IDataConnection adapter) =>
adapter.ReceivedCalls().Any(c => c.GetMethodInfo().Name == "ConnectAsync");
// ── The connection actor ──
///
/// The headline behaviour: a live, CONNECTED connection told to reconfigure builds a new
/// adapter from the new configuration and re-issues the subscription it already holds against
/// it — without the subscriber doing anything, and without the connection actor being stopped.
///
///
/// Fails before the fix: there was no reconfigure command at all, and the create the
/// Deployment Manager re-issued was dropped by the manager's create-only guard, so the old
/// adapter kept serving old-plc indefinitely.
///
///
[Fact]
public void Reconfigure_WhileConnected_SwapsTheAdapter_AndReSubscribesExistingTags()
{
const string name = "reconf-connected";
var oldAdapter = HealthyAdapter("sub-old");
var newAdapter = HealthyAdapter("sub-new");
_factory.Create("OpcUa", Arg.Is>(d => d["Endpoint"] == NewEndpoint))
.Returns(newAdapter);
var actor = CreateConnectionActor(name, oldAdapter, Config(OldEndpoint));
AwaitCondition(() => HasEnteredConnectedState(name), TimeSpan.FromSeconds(5),
TimeSpan.FromMilliseconds(10));
actor.Tell(new SubscribeTagsRequest(
"c1", "inst1", name, ["sensor/temp"], DateTimeOffset.UtcNow));
ExpectMsg(TimeSpan.FromSeconds(5));
AwaitCondition(() => Subscribed(oldAdapter), TimeSpan.FromSeconds(5));
actor.Tell(new ReconfigureConnectionCommand(
name, "OpcUa", Config(NewEndpoint), null, 3));
AwaitCondition(() => CreatedAdapterFor(NewEndpoint), TimeSpan.FromSeconds(5),
TimeSpan.FromMilliseconds(10));
AwaitCondition(() => Connected(newAdapter), TimeSpan.FromSeconds(5));
// The whole point: the subscription survives the swap and is re-issued for free.
AwaitCondition(() => Subscribed(newAdapter), TimeSpan.FromSeconds(5),
TimeSpan.FromMilliseconds(10));
}
///
/// The adapter generation must be bumped by the swap, exactly as it is on failover: the
/// outgoing adapter's subscription callbacks captured Self and keep firing, and a value from
/// the endpoint the operator has just corrected AWAY from must never be delivered as if it
/// came from the new one.
///
[Fact]
public void Reconfigure_DropsValuesStillArrivingFromTheOldAdapter()
{
const string name = "reconf-generation";
SubscriptionCallback? oldCallback = null;
var oldAdapter = HealthyAdapter("sub-old");
oldAdapter.SubscribeAsync("sensor/temp", Arg.Any(), Arg.Any())
.Returns(ci =>
{
oldCallback = (SubscriptionCallback)ci[1];
return Task.FromResult("sub-old");
});
var newAdapter = HealthyAdapter("sub-new");
_factory.Create("OpcUa", Arg.Is>(d => d["Endpoint"] == NewEndpoint))
.Returns(newAdapter);
var actor = CreateConnectionActor(name, oldAdapter, Config(OldEndpoint));
AwaitCondition(() => HasEnteredConnectedState(name), TimeSpan.FromSeconds(5),
TimeSpan.FromMilliseconds(10));
actor.Tell(new SubscribeTagsRequest(
"c1", TestActor.Path.Name, name, ["sensor/temp"], DateTimeOffset.UtcNow));
ExpectMsg(TimeSpan.FromSeconds(5));
AwaitCondition(() => oldCallback != null, TimeSpan.FromSeconds(5));
actor.Tell(new ReconfigureConnectionCommand(
name, "OpcUa", Config(NewEndpoint), null, 3));
// The swap flips every subscribed tag to bad quality, same as a disconnect.
ExpectMsg(TimeSpan.FromSeconds(5));
AwaitCondition(() => Subscribed(newAdapter), TimeSpan.FromSeconds(5));
ExpectNoMsg(TimeSpan.FromMilliseconds(300));
oldCallback!("sensor/temp", new TagValue(999, QualityCode.Good, DateTimeOffset.UtcNow));
ExpectNoMsg(TimeSpan.FromSeconds(1));
}
///
/// The state that matters most, and the one a stash would break. A connection whose
/// configured endpoint is wrong or dead never reaches Connected at all — it loops in the
/// Connecting state retrying — which is EXACTLY when a corrected configuration is sent. If
/// the reconfigure were stashed with the subscribe/write requests it would wait for the old
/// (broken) configuration to succeed first, i.e. forever.
///
[Fact]
public void Reconfigure_WhileStillConnecting_IsHonoured_NotStashed()
{
const string name = "reconf-connecting";
var deadAdapter = Substitute.For();
deadAdapter.ConnectAsync(Arg.Any>(), Arg.Any())
.Returns(Task.FromException(new Exception("Connection refused")));
var newAdapter = HealthyAdapter("sub-new");
_factory.Create("OpcUa", Arg.Is>(d => d["Endpoint"] == NewEndpoint))
.Returns(newAdapter);
var actor = CreateConnectionActor(name, deadAdapter, Config(OldEndpoint));
AwaitCondition(() => Connected(deadAdapter), TimeSpan.FromSeconds(5));
Assert.False(HasEnteredConnectedState(name));
actor.Tell(new ReconfigureConnectionCommand(
name, "OpcUa", Config(NewEndpoint), null, 3));
AwaitCondition(() => CreatedAdapterFor(NewEndpoint), TimeSpan.FromSeconds(5),
TimeSpan.FromMilliseconds(10));
AwaitCondition(() => HasEnteredConnectedState(name), TimeSpan.FromSeconds(5),
TimeSpan.FromMilliseconds(10));
}
///
/// The same, one state over: a connection that HAD been up and is now looping in the
/// Reconnecting state. Reconnecting stashes subscribes and writes too, so the reconfigure
/// must be handled there as well — and the subscriptions it re-issues are the ones the
/// registry has been holding across the outage.
///
[Fact]
public void Reconfigure_WhileReconnecting_IsHonoured_AndReSubscribes()
{
const string name = "reconf-reconnecting";
var connects = 0;
var oldAdapter = HealthyAdapter("sub-old");
oldAdapter.ConnectAsync(Arg.Any>(), Arg.Any())
.Returns(_ => Interlocked.Increment(ref connects) == 1
? Task.CompletedTask
: Task.FromException(new Exception("still down")));
var newAdapter = HealthyAdapter("sub-new");
_factory.Create("OpcUa", Arg.Is>(d => d["Endpoint"] == NewEndpoint))
.Returns(newAdapter);
var actor = CreateConnectionActor(name, oldAdapter, Config(OldEndpoint));
AwaitCondition(() => HasEnteredConnectedState(name), TimeSpan.FromSeconds(5),
TimeSpan.FromMilliseconds(10));
actor.Tell(new SubscribeTagsRequest(
"c1", "inst1", name, ["sensor/temp"], DateTimeOffset.UtcNow));
ExpectMsg(TimeSpan.FromSeconds(5));
// Drop the link; with no backup configured it retries the same dead endpoint forever.
RaiseDisconnected(oldAdapter);
AwaitCondition(() => connects >= 2, TimeSpan.FromSeconds(5));
actor.Tell(new ReconfigureConnectionCommand(
name, "OpcUa", Config(NewEndpoint), null, 3));
AwaitCondition(() => CreatedAdapterFor(NewEndpoint), TimeSpan.FromSeconds(5),
TimeSpan.FromMilliseconds(10));
AwaitCondition(() => Subscribed(newAdapter), TimeSpan.FromSeconds(5),
TimeSpan.FromMilliseconds(10));
}
///
/// A reconfigure carrying the configuration already in force must cost nothing. The sender is
/// hash-gated and does not send one, but an adapter swap drops the link and re-subscribes
/// every tag, so a redundant command must never be able to do that.
///
[Fact]
public void Reconfigure_WithTheConfigurationAlreadyInForce_IsANoOp()
{
const string name = "reconf-identical";
var adapter = HealthyAdapter("sub-old");
var actor = CreateConnectionActor(name, adapter, Config(OldEndpoint));
AwaitCondition(() => HasEnteredConnectedState(name), TimeSpan.FromSeconds(5),
TimeSpan.FromMilliseconds(10));
actor.Tell(new ReconfigureConnectionCommand(
name, "OpcUa", Config(OldEndpoint), null, 3));
// No adapter is ever built: the factory is only consulted by a swap.
ExpectNoMsg(TimeSpan.FromMilliseconds(500));
Assert.False(CreatedAdapterFor(OldEndpoint));
Assert.DoesNotContain(_factory.ReceivedCalls(),
c => c.GetMethodInfo().Name == nameof(IDataConnectionFactory.Create));
}
// ── The manager's routing ──
///
/// The manager forwards a reconfigure to the connection actor that owns the name — which is
/// the whole difference from , whose handler warns and
/// returns for a name it already holds.
///
[Fact]
public void Manager_Reconfigure_ForALiveConnection_ReachesTheConnectionActor()
{
var oldAdapter = HealthyAdapter("sub-old");
var newAdapter = HealthyAdapter("sub-new");
_factory.Create("OpcUa", Arg.Is>(d => d["Endpoint"] == OldEndpoint))
.Returns(oldAdapter);
_factory.Create("OpcUa", Arg.Is>(d => d["Endpoint"] == NewEndpoint))
.Returns(newAdapter);
var manager = Sys.ActorOf(Props.Create(() =>
new DataConnectionManagerActor(_factory, _options, _healthCollector)));
manager.Tell(new CreateConnectionCommand("conn1", "OpcUa", Config(OldEndpoint), null, 3));
AwaitCondition(() => Connected(oldAdapter), TimeSpan.FromSeconds(10));
manager.Tell(new ReconfigureConnectionCommand("conn1", "OpcUa", Config(NewEndpoint), null, 3));
AwaitCondition(() => Connected(newAdapter), TimeSpan.FromSeconds(10),
TimeSpan.FromMilliseconds(10));
}
///
/// A reconfigure for a name the manager does not hold falls back to a create, so the command
/// is safe whatever the manager's state — a sender whose view of which connections exist is
/// stale still converges on the intended configuration in one message rather than leaving
/// the connection missing until something else happens to create it.
///
[Fact]
public void Manager_Reconfigure_ForAnUnknownConnection_FallsBackToCreate()
{
var adapter = HealthyAdapter("sub-new");
_factory.Create("OpcUa", Arg.Is>(d => d["Endpoint"] == NewEndpoint))
.Returns(adapter);
var manager = Sys.ActorOf(Props.Create(() =>
new DataConnectionManagerActor(_factory, _options, _healthCollector)));
manager.Tell(new ReconfigureConnectionCommand("never-created", "OpcUa", Config(NewEndpoint), null, 3));
AwaitCondition(() => Connected(adapter), TimeSpan.FromSeconds(10),
TimeSpan.FromMilliseconds(10));
// And it is a real, routable connection afterwards — not an orphan.
manager.Tell(new SubscribeTagsRequest(
"c1", "inst1", "never-created", ["sensor/temp"], DateTimeOffset.UtcNow));
var response = ExpectMsg(TimeSpan.FromSeconds(10));
Assert.True(response.Success);
}
}