diff --git a/src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Actors/DataConnectionActor.cs b/src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Actors/DataConnectionActor.cs index c5a16c11..d3f371bb 100644 --- a/src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Actors/DataConnectionActor.cs +++ b/src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Actors/DataConnectionActor.cs @@ -234,6 +234,51 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers /// private IActorRef _self = null!; + /// + /// The delegate currently hooked to _adapter.Disconnected. It closes over the + /// in force when it was hooked, so the resulting + /// carries the generation of the adapter that raised it — + /// a method group could not. Kept in a field so and + /// unhook exactly the instance they hooked. + /// + private Action? _disconnectedHandler; + + /// + /// Arch-review 2026-08-19 P2-6. Set when an for the + /// CURRENT adapter generation is received while the actor is still Connecting/Reconnecting — + /// i.e. the adapter connected and dropped again before the piped + /// for that attempt reached the mailbox. + /// + /// + /// Neither state used to handle the message at all, so it fell through to + /// Unhandled and the subsequent ConnectResult(true) declared the connection up + /// on an adapter that was already dead. Nothing recovers from there: an adapter fires + /// Disconnected at most once per successful connect (OpcUaDataConnection's + /// once-only _disconnectFired guard, reset only by the next successful + /// ConnectAsync), so a connection with zero subscriptions — nothing to fail at + /// connection level and drive the reconnect machine — sat Connected on a dead adapter until + /// the process restarted. + /// + /// + /// + /// It is cleared at the top of , so it only ever refers to + /// the attempt currently in flight: a disconnect belonging to a session that ended BEFORE + /// this attempt began (e.g. the adapter's Disconnected trailing a connection-level + /// subscribe failure that already drove the actor into Reconnecting) is discarded rather than + /// replayed against the fresh session. That is also why the flag is used instead of + /// Stash: the stash has no selective drop, so a stashed disconnect would be replayed + /// after the next successful connect and cost a spurious reconnect cycle. + /// + /// + /// + /// Residual (accepted, and in the safe direction): a disconnect raised by the PREVIOUS + /// session but delivered after the next attempt has already cleared the flag is attributed to + /// the new attempt, costing one extra reconnect cycle. Being wedged Connected on a dead + /// adapter is the failure this trades against. + /// + /// + private bool _disconnectDuringConnectAttempt; + /// /// Initializes the data connection actor with its adapter and configuration. /// @@ -283,23 +328,43 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers _self = Self; // Listen for unexpected adapter disconnections - _adapter.Disconnected += OnAdapterDisconnected; + HookAdapterDisconnected(); BecomeConnecting(); } - private void OnAdapterDisconnected() + /// + /// Subscribes to the CURRENT adapter's Disconnected event with a delegate that closes + /// over the generation in force right now, so the marshalled + /// identifies the adapter that raised it. Callers must + /// have bumped for the new adapter first + /// (see ). + /// + private void HookAdapterDisconnected() { - // Marshal the event onto the actor's message loop using captured _self reference. - // This runs on a background thread (gRPC stream reader), so Self would throw. - _self.Tell(new AdapterDisconnected()); + var generation = _adapterGeneration; + var self = _self; + // Marshal the event onto the actor's message loop using the captured _self reference. + // This runs on a background thread (keep-alive / gRPC stream reader), so Self would throw. + _disconnectedHandler = () => self.Tell(new AdapterDisconnected(generation)); + _adapter.Disconnected += _disconnectedHandler; + } + + /// Unhooks the delegate installed by , if any. + private void UnhookAdapterDisconnected() + { + if (_disconnectedHandler is null) + return; + + _adapter.Disconnected -= _disconnectedHandler; + _disconnectedHandler = null; } /// protected override void PostStop() { _log.Info("DataConnectionActor [{0}] stopping — disposing adapter", _connectionName); - _adapter.Disconnected -= OnAdapterDisconnected; + UnhookAdapterDisconnected(); // S9: fire-and-forget, but observe a faulted dispose so a failing adapter // teardown is logged rather than swallowed into an unobserved task exception. _adapter.DisposeAsync().AsTask().ContinueWith( @@ -340,6 +405,14 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers case ConnectResult result: HandleConnectResult(result); break; + case AdapterDisconnected disconnected: + // P2-6 — the adapter connected and dropped again before this attempt's + // ConnectResult was applied. Remember it (see _disconnectDuringConnectAttempt) + // rather than acting now: acting would put the actor into Reconnecting and the + // in-flight ConnectResult(true) would then be applied THERE, re-subscribing and + // declaring the same dead adapter Connected anyway. + NotePendingDisconnect(disconnected); + break; case SubscribeTagsRequest: case WriteTagRequest: case WriteTagBatchRequest: @@ -467,8 +540,13 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers case QualityFlushTick: FlushQualityCountersIfDirty(); break; - case AdapterDisconnected: - HandleDisconnect(); + case AdapterDisconnected disconnected: + // P2-6 — fence the disconnect on its adapter generation. The outgoing adapter's + // event can fire concurrently with SwapAdapter's unhook, so a Disconnected from + // the endpoint we have just failed over (or reconfigured) AWAY from could arrive + // here and knock the healthy replacement back into a reconnect cycle. + if (!IsStaleDisconnect(disconnected)) + HandleDisconnect(); break; case ReconfigureConnectionCommand rc: HandleReconfigure(rc); @@ -600,14 +678,17 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers /// private void SwapAdapter(IDictionary newConfig) { - _adapter.Disconnected -= OnAdapterDisconnected; + UnhookAdapterDisconnected(); _ = _adapter.DisposeAsync().AsTask(); _adapter = _factory.Create(_protocolType, newConfig); _connectionDetails = newConfig; - _adapter.Disconnected += OnAdapterDisconnected; _adapterGeneration++; + // Hooked AFTER the bump so the new adapter's handler stamps the NEW generation; the + // outgoing adapter's handler was captured with the old one and is now recognised as + // stale by IsStaleDisconnect even if it fires concurrently with the unhook above. + HookAdapterDisconnected(); } /// @@ -702,6 +783,12 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers case ConnectResult result: HandleReconnectResult(result); break; + case AdapterDisconnected disconnected: + // P2-6 — same gap as Connecting: a reconnect that succeeds and drops again before + // its ConnectResult is applied would otherwise be declared Connected on a dead + // adapter that will never raise Disconnected a second time. + NotePendingDisconnect(disconnected); + break; case SubscribeTagsRequest: case WriteTagRequest: case WriteTagBatchRequest: @@ -777,6 +864,10 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers private void HandleAttemptConnect() { _log.Debug("[{0}] Attempting connection...", _connectionName); + // P2-6 — scope the pending-disconnect flag to THIS attempt. Anything noted before now + // belongs to a session that is already over and must not be replayed against the one + // about to start. + _disconnectDuringConnectAttempt = false; var self = Self; // Stamp the generation this attempt belongs to, exactly as every other piped // completion does. A failover or an in-place reconfigure landing while ConnectAsync @@ -827,6 +918,75 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers return true; } + /// + /// True when this was raised by an adapter that has since + /// been swapped out (failover or in-place reconfigure), in which case it is dropped — the + /// disconnect twin of (arch-review 2026-08-19 P2-6). + /// unhooks the outgoing adapter, but the event can already be + /// executing on a background thread when it does, so the message can still be delivered + /// after the swap. + /// + private bool IsStaleDisconnect(AdapterDisconnected disconnected) + { + if (disconnected.AdapterGeneration == _adapterGeneration) + return false; + + _log.Debug( + "[{0}] Dropping disconnect notification from adapter generation {1} (current {2}) — " + + "the adapter that raised it has been replaced", + _connectionName, disconnected.AdapterGeneration, _adapterGeneration); + return true; + } + + /// + /// Records a current-generation disconnect that arrived while a connect attempt was still in + /// flight, so / can + /// refuse to declare the connection up. See . + /// + private void NotePendingDisconnect(AdapterDisconnected disconnected) + { + if (IsStaleDisconnect(disconnected)) + return; + + _log.Warning( + "[{0}] Adapter reported a disconnect while a connect attempt was still in flight — " + + "the attempt's result will be treated as a failed connection", _connectionName); + _disconnectDuringConnectAttempt = true; + } + + /// + /// Reads and clears , logging the demotion. True + /// means the successful connect result being applied belongs to a session the adapter has + /// already reported as dropped. + /// + private bool ConsumeDisconnectDuringConnectAttempt() + { + if (!_disconnectDuringConnectAttempt) + return false; + + _disconnectDuringConnectAttempt = false; + _log.Warning( + "[{0}] Connect succeeded but the adapter had already reported the session lost — " + + "treating the attempt as a failed connect", _connectionName); + return true; + } + + /// + /// Applies a demoted connect success (see ) + /// exactly as a connect FAILURE: the session the "success" describes is already gone, so it + /// must count toward the failover budget and re-arm the retry timer rather than declare the + /// connection up. Verbatim the tail of both result handlers' failure branches, which is the + /// point — a flapping endpoint that drops inside every connect window still reaches its + /// backup instead of retrying the same dead primary forever. + /// + private void DemoteConnectAttemptToFailure() + { + _consecutiveFailures++; + CountFailureAndMaybeFailover( + "adapter reported the session lost before the connect result was applied"); + Timers.StartSingleTimer("reconnect", new AttemptConnect(), _options.ReconnectInterval); + } + private void HandleConnectResult(ConnectResult result) { if (IsStaleConnectResult(result)) @@ -834,6 +994,16 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers if (result.Success) { + // P2-6 — a disconnect that raced ahead of this result means the session this + // "success" describes is already gone; going Connected here would strand the + // connection on a dead adapter that never raises Disconnected again. Stay in + // Connecting and retry, exactly as a failed connect would. + if (ConsumeDisconnectDuringConnectAttempt()) + { + DemoteConnectAttemptToFailure(); + return; + } + _log.Info("[{0}] Connection established", _connectionName); BecomeConnected(); } @@ -858,6 +1028,15 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers if (result.Success) { + // P2-6 — same demotion as the initial connect: the reconnect landed but the adapter + // has already reported the new session lost, so re-subscribing and going Connected + // would wedge on a dead adapter. Stay in Reconnecting and retry. + if (ConsumeDisconnectDuringConnectAttempt()) + { + DemoteConnectAttemptToFailure(); + return; + } + _log.Info("[{0}] Reconnected successfully on {1} endpoint", _connectionName, _activeEndpoint); _consecutiveFailures = 0; @@ -2131,10 +2310,79 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers })); } + // NOTE: `self` is the Self reference captured on the actor thread above — this runs on + // a background task, where Akka's Self property is invalid. + SeedNewlyResolvedTags(self, adapter, generation, rows); + return new BatchSubscribeCompleted(rows, generation, BatchSubscribeSource.ResolutionProbe); }).PipeTo(self); } + /// + /// Reads the current value of every tag this probe round has just resolved and delivers it + /// through the normal generation-guarded path — the same seed + /// the initial subscribe () and the reconnect re-subscribe + /// () already perform, which the resolution probe was missing + /// (arch-review 2026-08-19 P2-5). + /// + /// + /// A newly advised subscription only starts producing values on the source's next CHANGE. + /// OPC UA masks that with an initial data-change notification on subscription creation, but + /// an MxGateway advise can legitimately return VT_EMPTY, so a tag that resolved LATE + /// (device still booting, item created after deployment) and then never changes had no value + /// path at all — its attribute stayed Uncertain for the life of the connection, healed only + /// by an unrelated reconnect. + /// + /// + /// + /// Unlike , the seeds are delivered directly rather than carried + /// back on the completion message: the probe only ever covers tags already present in + /// / , so the + /// registration-ordering hazard that forces the initial subscribe to defer does not exist + /// here. It runs as its own fire-and-forget task (mirroring the reconnect re-seed) so the + /// seed reads — bounded by SeedOverallTimeout, not by the subscribe — never delay the + /// that carries the round's resolution bookkeeping and + /// re-arms the retry timer. Values that arrive after a failover or a drop to Reconnecting are + /// discarded by the generation guard / state handler exactly like any other tag value. + /// + /// + /// + /// The actor's own reference, captured on the actor thread by the caller — this method is + /// invoked from a background task, where Akka's Self property is invalid. + /// + /// The adapter the probe round ran against. + /// The adapter generation the probe round was stamped with. + /// The probe round's per-tag subscribe results. + private void SeedNewlyResolvedTags( + IActorRef self, IDataConnection adapter, int generation, IReadOnlyList rows) + { + var resolved = new List(rows.Count); + foreach (var row in rows) + { + if (row is { Success: true, SubscriptionId: not null }) + resolved.Add(row.TagPath); + } + + if (resolved.Count == 0) + return; + + _ = Task.Run(async () => + { + try + { + var seeds = await SeedTagsAsync(adapter, resolved); + foreach (var seed in seeds) + self.Tell(new TagValueReceived(seed.TagPath, seed.Value, generation)); + } + catch (Exception ex) + { + // Fire-and-forget: guarantee a trace rather than an unobserved task fault. + // SeedTagsAsync already swallows per-read errors, so reaching here is unexpected. + _log.Warning("[{0}] Tag-resolution re-seed task faulted: {1}", _connectionName, ex.Message); + } + }); + } + /// /// Arms the single-shot tag-resolution retry timer at the current backoff interval /// when unresolved tags remain, and cancels it when none do. Single-shot (rescheduled @@ -2961,7 +3209,12 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers /// rather than applied to its replacement (see ). /// internal record ConnectResult(bool Success, string? Error, int AdapterGeneration); - internal record AdapterDisconnected; + /// + /// The adapter's Disconnected event marshalled onto the actor thread, stamped with the + /// the raising adapter was hooked at so a notification from a + /// swapped-out adapter can be dropped (). + /// + internal record AdapterDisconnected(int AdapterGeneration); internal record TagValueReceived(string TagPath, TagValue Value, int AdapterGeneration); internal record RetryTagResolution; diff --git a/tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Actors/DataConnectionDisconnectRaceTests.cs b/tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Actors/DataConnectionDisconnectRaceTests.cs new file mode 100644 index 00000000..f8b8e809 --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Actors/DataConnectionDisconnectRaceTests.cs @@ -0,0 +1,242 @@ +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)); + } +} diff --git a/tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Actors/DataConnectionProbeSeedTests.cs b/tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Actors/DataConnectionProbeSeedTests.cs new file mode 100644 index 00000000..fd55b849 --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Actors/DataConnectionProbeSeedTests.cs @@ -0,0 +1,106 @@ +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-5 — the tag-resolution PROBE path did not seed the tags it +/// resolved. +/// +/// +/// The gap this pins. Both other subscribe paths read each newly advised tag's current +/// value and deliver it (HandleSubscribe carries the seeds back on SubscribeCompleted; +/// ReSubscribeAll re-seeds after a reconnect), because a fresh advise only produces a value +/// on the source's next CHANGE. HandleRetryTagResolution did not. OPC UA masks that with an +/// initial data-change notification on subscription creation, but an MxGateway advise can +/// legitimately return VT_EMPTY — so a STATIC tag that resolved LATE (device still booting, +/// item created after deployment) had no value path at all and its attribute stayed Uncertain for +/// the life of the connection. +/// +/// +[Collection("DataConnectionManagerActor")] +public class DataConnectionProbeSeedTests : TestKit +{ + private const string Tag = "plant/static/setpoint"; + + private readonly ISiteHealthCollector _health = Substitute.For(); + private readonly IDataConnectionFactory _factory = Substitute.For(); + + private readonly DataConnectionOptions _options = new() + { + ReconnectInterval = TimeSpan.FromMilliseconds(100), + TagResolutionRetryInterval = TimeSpan.FromMilliseconds(150), + WriteTimeout = TimeSpan.FromSeconds(5), + SeedReadMaxAttempts = 1, + SeedReadRetryDelay = TimeSpan.FromMilliseconds(10) + }; + + private IActorRef CreateConnectionActor(string name, IDataConnection adapter) => + Sys.ActorOf(Props.Create(() => + new DataConnectionActor(name, adapter, _options, _health, _factory, "OpcUa")), name); + + private bool HasEnteredConnectedState(string connectionName) => + _health.ReceivedCalls().Any(c => + c.GetMethodInfo().Name == nameof(ISiteHealthCollector.UpdateConnectionHealth) && + c.GetArguments() is [string name, ConnectionHealth.Connected, ..] && + name == connectionName); + + /// + /// A tag that fails to resolve on the initial subscribe, resolves on the first probe round, + /// and then never changes again must still reach its subscriber with a Good value. + /// + /// + /// The adapter here models exactly the MxGateway shape the finding calls out: the advise + /// succeeds on the retry but the subscription callback is NEVER invoked (a VT_EMPTY + /// advise on a static item), so the seed read is the tag's only possible value source. + /// + /// + /// Fails before the fix: the probe resolved the tag and stopped there — the + /// subscriber received only the Bad-quality resolution-failure push from the initial + /// subscribe, and this fish for a Good value timed out. + /// + [Fact] + public void ProbeResolvedTag_IsSeeded_SoAStaticTagLeavesUncertain() + { + const string name = "probe-seed"; + var subscribeCalls = 0; + + var adapter = Substitute.For(); + adapter.ConnectAsync(Arg.Any>(), Arg.Any()) + .Returns(Task.CompletedTask); + adapter.Status.Returns(ConnectionHealth.Connected); + adapter.SubscribeAsync(Tag, Arg.Any(), Arg.Any()) + .Returns(_ => Interlocked.Increment(ref subscribeCalls) == 1 + // Initial subscribe: the device is still booting — a genuine (non + // connection-level) resolution failure, so the tag lands in _unresolvedTags. + ? Task.FromException(new KeyNotFoundException("node not found yet")) + // Probe round: the advise succeeds. The callback is deliberately never invoked. + : Task.FromResult("sub-1")); + adapter.ReadAsync(Tag, Arg.Any()) + .Returns(new ReadResult(true, new TagValue(42, QualityCode.Good, DateTimeOffset.UtcNow), null)); + + var actor = CreateConnectionActor(name, adapter); + AwaitCondition(() => HasEnteredConnectedState(name), + TimeSpan.FromSeconds(5), TimeSpan.FromMilliseconds(10)); + + var subscriber = CreateTestProbe(); + subscriber.Send(actor, new SubscribeTagsRequest("c1", "inst1", name, [Tag], DateTimeOffset.UtcNow)); + + // The initial subscribe fails: Bad-quality push, then the ack. + subscriber.FishForMessage(_ => true, TimeSpan.FromSeconds(5)); + + // The probe round resolves the tag; the seed read is the only thing that can now + // deliver a value, because the advise callback never fires for this static item. + var seeded = subscriber.FishForMessage( + m => m.TagPath == Tag && m.Quality == QualityCode.Good, TimeSpan.FromSeconds(10)); + + Assert.Equal(42, seeded.Value); + Assert.True(Volatile.Read(ref subscribeCalls) >= 2, "the probe round must have re-subscribed the tag"); + } +}