diff --git a/src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Actors/DataConnectionActor.cs b/src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Actors/DataConnectionActor.cs index bd5b316b..c5bb81d0 100644 --- a/src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Actors/DataConnectionActor.cs +++ b/src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Actors/DataConnectionActor.cs @@ -983,7 +983,7 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers // dropped, and logs any tag that never yields a value. var seedValues = await SeedTagsAsync(adapter, tagsToSeed); - return new SubscribeCompleted(request, sender, results, seedValues); + return new SubscribeCompleted(request, sender, results, seedValues, generation, adapter); }).PipeTo(self); } @@ -1086,6 +1086,30 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers return Task.WhenAll(subscriptionIds.Select(id => adapter.UnsubscribeAsync(id))); } + /// + /// Best-effort release of handles created by a subscribe whose adapter has since been + /// swapped out. must target the adapter that CREATED the handles + /// — they are adapter-scoped (MxGateway's are plain ints that collide across sessions), so + /// releasing by handle against the replacement adapter can tear down a live subscription. + /// The outgoing adapter is disposed fire-and-forget by and may + /// already be gone, so a failure here is expected and logged at Debug rather than surfaced. + /// + private void ReleaseStaleAgainstOwnAdapter(string description, Func release) + { + try + { + release().ContinueWith( + t => _log.Debug("[{0}] Best-effort release of {1} failed: {2}", + _connectionName, description, t.Exception?.GetBaseException().Message), + TaskContinuationOptions.OnlyOnFaulted); + } + catch (Exception ex) + { + _log.Debug("[{0}] Best-effort release of {1} threw: {2}", + _connectionName, description, ex.Message); + } + } + /// /// Reads the current value of each tag so the Instance Actor /// has an initial value, retrying the still-empty subset a bounded number of times. A @@ -1256,6 +1280,51 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers private bool HandleSubscribeCompleted(SubscribeCompleted msg) { var instanceName = msg.Request.InstanceUniqueName; + + // Results produced against an adapter that has since been swapped out (failover or + // in-place reconfigure) are dropped WHOLE, mirroring the BatchSubscribeCompleted and + // TagValueReceived generation guards. An initial subscribe can take up to + // SeedOverallTimeout to complete, so this window is wide. Applying a stale completion + // stores a dead session's handle under the tag's key, which then makes the LIVE + // completion for the same tag look like a duplicate and get RELEASED (the tag reads + // resolved forever and delivers nothing); re-stamps the old endpoint's seed values with + // the CURRENT generation, so they bypass the stale-data guard and present as fresh; and + // lets a connection-level failure row from the dead adapter knock the healthy new + // connection back into Reconnecting. + if (msg.AdapterGeneration != _adapterGeneration) + { + var staleIds = msg.Results + .Where(r => r is { Success: true, AlreadySubscribed: false, SubscriptionId: not null }) + .Select(r => r.SubscriptionId!) + .ToList(); + + _log.Debug( + "[{0}] Dropping stale subscribe completion for instance {1} from adapter " + + "generation {2} (current {3}); releasing {4} handle(s) against the adapter that created them", + _connectionName, instanceName, msg.AdapterGeneration, _adapterGeneration, staleIds.Count); + + // Release against msg.Adapter, NEVER _adapter: handles are adapter-scoped and + // MxGateway's are plain ints that collide across sessions, so releasing by handle + // against the replacement adapter can tear down a live subscription. + ReleaseStaleAgainstOwnAdapter( + $"{staleIds.Count} tag subscription handle(s) from adapter generation {msg.AdapterGeneration}", + () => UnsubscribeIdsAsync(msg.Adapter, staleIds)); + + // _subscribesInFlight is deliberately NOT cleared here: ReSubscribeAll empties it + // wholesale once the replacement adapter connects, and a marker the new generation + // has since placed for the same tag must not be removed by this dead message. + + // Reply failure so the requester re-sends. Nothing has registered this request's + // tags in _subscriptionsByInstance (registration happens below, on the applied + // path), so ReSubscribeAll cannot pick them up — the InstanceActor's per-connection + // retry timer, armed until a Success arrives, is what re-establishes them. + msg.ReplyTo.Tell(new SubscribeTagsResponse( + msg.Request.CorrelationId, instanceName, false, + "adapter replaced while the subscribe was in flight — will retry", + DateTimeOffset.UtcNow)); + return false; + } + if (!_subscriptionsByInstance.TryGetValue(instanceName, out var instanceTags)) { // The instance was unsubscribed while the @@ -1292,7 +1361,10 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers } } - _ = UnsubscribeIdsAsync(_adapter, orphanedIds); + // Against the adapter that created them (identical to _adapter here — the + // generation guard above already proved the completion is current — but stating + // the ownership rule at every release site is what keeps it true). + _ = UnsubscribeIdsAsync(msg.Adapter, orphanedIds); // The original sender is already gone (unsubscribed). Telling a dead // ref produces a dead letter, which is the harmless and observable @@ -1400,8 +1472,9 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers } } - // Fire-and-forget release of every handle this request created redundantly. - _ = UnsubscribeIdsAsync(_adapter, redundantIds); + // Fire-and-forget release of every handle this request created redundantly, against the + // adapter that created them (see the orphaned-release site above). + _ = UnsubscribeIdsAsync(msg.Adapter, redundantIds); // Publish the resolution counts this request just changed. Without this push a // connection whose tags ALL resolve on the first attempt reported 0/0 forever: @@ -2502,14 +2575,46 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers alarmable.SubscribeAlarmsAsync(sourceRef, filter, t => self.Tell(new AlarmTransitionReceived(t, generation))) .ContinueWith(task => task.IsCompletedSuccessfully - ? new AlarmSubscribeCompleted(sourceRef, true, task.Result, null, subscriber, corr, inst) as object + ? new AlarmSubscribeCompleted( + sourceRef, true, task.Result, null, subscriber, corr, inst, generation, alarmable) as object : new AlarmSubscribeCompleted(sourceRef, false, null, - task.Exception?.GetBaseException().Message ?? "Unknown error", subscriber, corr, inst)) + task.Exception?.GetBaseException().Message ?? "Unknown error", + subscriber, corr, inst, generation, alarmable)) .PipeTo(self); } private void HandleAlarmSubscribeCompleted(AlarmSubscribeCompleted msg) { + // The alarm twin of HandleSubscribeCompleted's generation guard: a completion from an + // adapter that has since been swapped out is dropped whole. Applying it would store a + // dead feed's id — or, when the new adapter's feed is already stored, take the + // duplicate branch and release the LIVE feed, leaving the source mirrored but silent. + if (msg.AdapterGeneration != _adapterGeneration) + { + _log.Debug( + "[{0}] Dropping stale alarm subscribe completion for source {1} from adapter " + + "generation {2} (current {3})", + _connectionName, msg.SourceReference, msg.AdapterGeneration, _adapterGeneration); + + // Release the orphaned feed against the adapter that created it, never the current + // one (see ReleaseStaleAgainstOwnAdapter). + if (msg is { Success: true, SubscriptionId: not null }) + ReleaseStaleAgainstOwnAdapter( + $"the alarm feed for {msg.SourceReference} from adapter generation {msg.AdapterGeneration}", + () => msg.Adapter.UnsubscribeAlarmsAsync(msg.SubscriptionId)); + + // _alarmSubscribesInFlight is left untouched — ReSubscribeAllAlarms clears it + // wholesale and may already have re-registered this source against the new adapter. + // The subscriber state (_alarmSourceSubscribers) survives the swap, so the feed is + // re-established by ReSubscribeAllAlarms; the failure reply just lets the + // NativeAlarmActor retry rather than wait out its response timeout. + msg.ReplyTo?.Tell(new SubscribeAlarmsResponse( + msg.CorrelationId ?? string.Empty, msg.InstanceUniqueName ?? string.Empty, + false, "adapter replaced while the alarm subscribe was in flight — will retry", + DateTimeOffset.UtcNow)); + return; + } + _alarmSubscribesInFlight.Remove(msg.SourceReference); // The last (or only) subscriber may have been @@ -2521,15 +2626,16 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers // transitions to nobody for the lifetime of the adapter. if (!_alarmSourceSubscribers.ContainsKey(msg.SourceReference)) { - if (msg.Success && msg.SubscriptionId != null && - _adapter is IAlarmSubscribableConnection alarmable) + if (msg.Success && msg.SubscriptionId != null) { _log.Warning( "[{0}] AlarmSubscribeCompleted arrived for source {1} but the last " + "subscriber unsubscribed while the subscribe was in flight; releasing " + "the orphaned adapter alarm feed.", _connectionName, msg.SourceReference); - _ = alarmable.UnsubscribeAlarmsAsync(msg.SubscriptionId); + // Against the adapter that created the feed (identical to _adapter here — the + // generation guard above already proved the completion is current). + _ = msg.Adapter.UnsubscribeAlarmsAsync(msg.SubscriptionId); } // No live requester remains to receive a response. @@ -2547,14 +2653,11 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers // the stored id and leaking the already-tracked subscription. if (_alarmSubscriptionIds.ContainsKey(msg.SourceReference)) { - if (_adapter is IAlarmSubscribableConnection alarmable) - { - _log.Warning( - "[{0}] Duplicate alarm feed for source {1}; releasing the redundant " + - "subscription instead of overwriting the stored one.", - _connectionName, msg.SourceReference); - _ = alarmable.UnsubscribeAlarmsAsync(msg.SubscriptionId); - } + _log.Warning( + "[{0}] Duplicate alarm feed for source {1}; releasing the redundant " + + "subscription instead of overwriting the stored one.", + _connectionName, msg.SourceReference); + _ = msg.Adapter.UnsubscribeAlarmsAsync(msg.SubscriptionId); } else { @@ -2755,9 +2858,11 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers alarmable.SubscribeAlarmsAsync(sr, filter, t => self.Tell(new AlarmTransitionReceived(t, generation))) .ContinueWith(task => task.IsCompletedSuccessfully - ? new AlarmSubscribeCompleted(sr, true, task.Result, null, null, null, null) as object + ? new AlarmSubscribeCompleted( + sr, true, task.Result, null, null, null, null, generation, alarmable) as object : new AlarmSubscribeCompleted(sr, false, null, - task.Exception?.GetBaseException().Message ?? "Unknown error", null, null, null)) + task.Exception?.GetBaseException().Message ?? "Unknown error", + null, null, null, generation, alarmable)) .PipeTo(self); } } @@ -2803,16 +2908,30 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers internal record SubscribeTagResult( string TagPath, bool AlreadySubscribed, bool Success, string? SubscriptionId, string? Error, bool ConnectionLevelFailure = false); + + /// + /// Per-tag results of the initial subscribe issued off the actor thread by + /// . Carries the adapter generation it ran against — and the + /// adapter INSTANCE that produced the handles, because a stale completion's handles must be + /// released against their own adapter (see ). + /// internal record SubscribeCompleted( SubscribeTagsRequest Request, IActorRef ReplyTo, IReadOnlyList Results, - IReadOnlyList SeedValues); + IReadOnlyList SeedValues, int AdapterGeneration, IDataConnection Adapter); /// An initial-read value captured during subscribe, delivered after the /// instance's tags are registered for fan-out. internal record SeededValue(string TagPath, TagValue Value); internal record AlarmTransitionReceived(NativeAlarmTransition Transition, int AdapterGeneration); + + /// + /// Result of an alarm-feed subscribe issued off the actor thread. Carries the adapter + /// generation and the adapter instance for the same reason + /// does. + /// internal record AlarmSubscribeCompleted( string SourceReference, bool Success, string? SubscriptionId, string? Error, - IActorRef? ReplyTo, string? CorrelationId, string? InstanceUniqueName); + IActorRef? ReplyTo, string? CorrelationId, string? InstanceUniqueName, + int AdapterGeneration, IAlarmSubscribableConnection Adapter); public record GetHealthReport; } diff --git a/tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Actors/DataConnectionStaleCompletionTests.cs b/tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Actors/DataConnectionStaleCompletionTests.cs new file mode 100644 index 00000000..9b5193ce --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Actors/DataConnectionStaleCompletionTests.cs @@ -0,0 +1,360 @@ +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.Alarms; +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-1 — an in-flight subscribe that completes AFTER the adapter +/// has been swapped out (failover, or the in-place reconfigure of register row 97). +/// +/// +/// The gap these pin. SubscribeCompleted/AlarmSubscribeCompleted carried no +/// adapter generation, and their handlers applied them unconditionally — while an initial +/// subscribe can stay in flight for up to SeedOverallTimeout (120 s), so the window is +/// wide. A stale completion stored a DEAD session's handle under the tag's key, which then made +/// the live completion for the same tag look like a duplicate and get released: the tag reads +/// resolved forever and delivers nothing. It also re-stamped the old endpoint's seed values with +/// the CURRENT generation (bypassing the stale-data guard) and let a connection-level failure row +/// from the dead adapter knock the healthy new connection back into Reconnecting. The alarm twin +/// released the live alarm feed the same way. +/// +/// +/// +/// Handles are adapter-scoped — MxGateway's are plain ints that collide across sessions — so a +/// dropped completion's handles are released against the adapter that CREATED them, never the +/// current one. +/// +/// +[Collection("DataConnectionManagerActor")] +public class DataConnectionStaleCompletionTests : TestKit +{ + private const string OldEndpoint = "opc.tcp://old-plc:4840"; + private const string NewEndpoint = "opc.tcp://new-plc:4840"; + private const string Tag = "sensor/temp"; + private const string Source = "Tank01"; + + 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(OldEndpoint), null, 3)), name); + + private void ExpectNewAdapter(IDataConnection adapter) => + _factory.Create("OpcUa", Arg.Is>(d => d["Endpoint"] == NewEndpoint)) + .Returns(adapter); + + 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 IDataConnection HealthyAlarmAdapter(string alarmSubscriptionId) + { + var adapter = Substitute.For(); + adapter.ConnectAsync(Arg.Any>(), Arg.Any()) + .Returns(Task.CompletedTask); + adapter.Status.Returns(ConnectionHealth.Connected); + ((IAlarmSubscribableConnection)adapter) + .SubscribeAlarmsAsync(Arg.Any(), Arg.Any(), + Arg.Any(), Arg.Any()) + .Returns(Task.FromResult(alarmSubscriptionId)); + return adapter; + } + + private static NativeAlarmTransition Raise(string sourceRef, string sourceObj) => + new(sourceRef, sourceObj, "AnalogLimit.Hi", AlarmTransitionKind.Raise, + new AlarmConditionState(true, false, null, AlarmShelveState.Unshelved, false, 500), + "Process", "hi", "hi", "", "", null, DateTimeOffset.UtcNow, "92", "90"); + + 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); + + private static bool CalledWith(IDataConnection adapter, string method, string firstArg) => + adapter.ReceivedCalls().Any(c => + c.GetMethodInfo().Name == method && c.GetArguments()[0] as string == firstArg); + + private static bool Called(IDataConnection adapter, string method) => + adapter.ReceivedCalls().Any(c => c.GetMethodInfo().Name == method); + + // ── Tag subscribe ── + + /// + /// The headline data-loss case: the stale completion must not park its dead handle under the + /// tag's key, because the live completion for the same tag then reads as a duplicate and has + /// its (working) monitored item released — leaving the tag permanently resolved and silent. + /// + /// + /// Fails before the fix: the stale completion stored sub-old, and the live + /// handle sub-new was released against the new adapter. + /// + /// + [Fact] + public void StaleSubscribeCompleted_DoesNotStoreItsHandle_AndLeavesTheLiveMonitoredItemAlone() + { + const string name = "stale-sub-handle"; + var gate = new TaskCompletionSource(); + + var oldAdapter = HealthyAdapter("sub-old"); + oldAdapter.SubscribeAsync(Tag, Arg.Any(), Arg.Any()) + .Returns(_ => gate.Task); + var newAdapter = HealthyAdapter("sub-new"); + ExpectNewAdapter(newAdapter); + + var actor = CreateConnectionActor(name, oldAdapter); + AwaitCondition(() => HealthTransitions(name, ConnectionHealth.Connected) >= 1, + TimeSpan.FromSeconds(5), TimeSpan.FromMilliseconds(10)); + + var probe = CreateTestProbe(); + probe.Send(actor, new SubscribeTagsRequest("c1", "inst1", name, [Tag], DateTimeOffset.UtcNow)); + AwaitCondition(() => CalledWith(oldAdapter, nameof(IDataConnection.SubscribeAsync), Tag), + TimeSpan.FromSeconds(5)); + + actor.Tell(new ReconfigureConnectionCommand(name, "OpcUa", Config(NewEndpoint), null, 3)); + AwaitCondition(() => HealthTransitions(name, ConnectionHealth.Connected) >= 2, + TimeSpan.FromSeconds(5), TimeSpan.FromMilliseconds(10)); + + // The Instance Actor's retry — the live subscribe, against the new adapter. + probe.Send(actor, new SubscribeTagsRequest("c2", "inst1", name, [Tag], DateTimeOffset.UtcNow)); + probe.FishForMessage( + m => m.CorrelationId == "c2" && m.Success, TimeSpan.FromSeconds(5)); + + // Only now does the generation-N subscribe answer. + gate.SetResult("sub-old"); + AwaitCondition(() => CalledWith(oldAdapter, nameof(IDataConnection.UnsubscribeAsync), "sub-old"), + TimeSpan.FromSeconds(5), TimeSpan.FromMilliseconds(10)); + + Assert.False(Called(newAdapter, nameof(IDataConnection.UnsubscribeAsync))); + + // The registry still holds the LIVE handle: the teardown releases sub-new, not sub-old. + probe.Send(actor, new UnsubscribeTagsRequest("c3", "inst1", name, DateTimeOffset.UtcNow)); + AwaitCondition(() => CalledWith(newAdapter, nameof(IDataConnection.UnsubscribeAsync), "sub-new"), + TimeSpan.FromSeconds(5), TimeSpan.FromMilliseconds(10)); + Assert.False(CalledWith(newAdapter, nameof(IDataConnection.UnsubscribeAsync), "sub-old")); + } + + /// + /// The handles a dropped completion did create are released — against the adapter that + /// created them. Releasing them against the CURRENT adapter is not a harmless mismatch: + /// MxGateway subscription handles are plain ints reused across sessions, so the release + /// would land on whatever the new session happens to have under that number. + /// + [Fact] + public void StaleSubscribeCompleted_ReleasesItsHandles_AgainstTheAdapterThatCreatedThem() + { + const string name = "stale-sub-release"; + var gate = new TaskCompletionSource(); + + var oldAdapter = HealthyAdapter("sub-old"); + oldAdapter.SubscribeAsync(Tag, Arg.Any(), Arg.Any()) + .Returns(_ => gate.Task); + var newAdapter = HealthyAdapter("sub-new"); + ExpectNewAdapter(newAdapter); + + var actor = CreateConnectionActor(name, oldAdapter); + AwaitCondition(() => HealthTransitions(name, ConnectionHealth.Connected) >= 1, + TimeSpan.FromSeconds(5), TimeSpan.FromMilliseconds(10)); + + var probe = CreateTestProbe(); + probe.Send(actor, new SubscribeTagsRequest("c1", "inst1", name, [Tag], DateTimeOffset.UtcNow)); + AwaitCondition(() => CalledWith(oldAdapter, nameof(IDataConnection.SubscribeAsync), Tag), + TimeSpan.FromSeconds(5)); + + actor.Tell(new ReconfigureConnectionCommand(name, "OpcUa", Config(NewEndpoint), null, 3)); + AwaitCondition(() => HealthTransitions(name, ConnectionHealth.Connected) >= 2, + TimeSpan.FromSeconds(5), TimeSpan.FromMilliseconds(10)); + + gate.SetResult("sub-old"); + + AwaitCondition(() => CalledWith(oldAdapter, nameof(IDataConnection.UnsubscribeAsync), "sub-old"), + TimeSpan.FromSeconds(5), TimeSpan.FromMilliseconds(10)); + Assert.False(Called(newAdapter, nameof(IDataConnection.UnsubscribeAsync))); + + // The requester is told the subscribe did not take, so its retry timer re-sends: the + // dropped completion registered nothing, so the reconnect re-subscribe cannot cover it. + probe.FishForMessage( + m => m.CorrelationId == "c1" && !m.Success, TimeSpan.FromSeconds(5)); + } + + /// + /// The seed values a dropped completion carries are readings from the endpoint the operator + /// has just corrected AWAY from. Applying them re-stamped each one with the CURRENT adapter + /// generation, which is precisely what the generation guard on TagValueReceived + /// exists to stop — the old endpoint's process values presented as fresh. + /// + [Fact] + public void StaleSubscribeCompleted_DoesNotDeliverItsSeedValues() + { + const string name = "stale-sub-seed"; + var gate = new TaskCompletionSource(); + + var oldAdapter = HealthyAdapter("sub-old"); + oldAdapter.SubscribeAsync(Tag, Arg.Any(), Arg.Any()) + .Returns(_ => gate.Task); + oldAdapter.ReadAsync(Tag, Arg.Any()) + .Returns(new ReadResult(true, new TagValue(999, QualityCode.Good, DateTimeOffset.UtcNow), null)); + var newAdapter = HealthyAdapter("sub-new"); + ExpectNewAdapter(newAdapter); + + var actor = CreateConnectionActor(name, oldAdapter); + AwaitCondition(() => HealthTransitions(name, ConnectionHealth.Connected) >= 1, + TimeSpan.FromSeconds(5), TimeSpan.FromMilliseconds(10)); + + var probe = CreateTestProbe(); + probe.Send(actor, new SubscribeTagsRequest("c1", "inst1", name, [Tag], DateTimeOffset.UtcNow)); + AwaitCondition(() => CalledWith(oldAdapter, nameof(IDataConnection.SubscribeAsync), Tag), + TimeSpan.FromSeconds(5)); + + actor.Tell(new ReconfigureConnectionCommand(name, "OpcUa", Config(NewEndpoint), null, 3)); + AwaitCondition(() => HealthTransitions(name, ConnectionHealth.Connected) >= 2, + TimeSpan.FromSeconds(5), TimeSpan.FromMilliseconds(10)); + + gate.SetResult("sub-old"); + + var received = probe.ReceiveWhile(TimeSpan.FromSeconds(2), m => m); + + // Positive control: the completion really did arrive and was dropped, rather than the + // window simply having been too short to observe anything. + Assert.Contains(received, m => m is SubscribeTagsResponse { Success: false }); + Assert.DoesNotContain(received, m => m is TagValueUpdate u && Equals(u.Value, 999)); + } + + /// + /// A connection-level failure row belongs to the adapter that produced it. Acting on a stale + /// one drives the healthy replacement connection into Reconnecting — dropping every tag to + /// bad quality and re-subscribing the whole registry for nothing. + /// + [Fact] + public void StaleConnectionLevelFailure_DoesNotDriveTheNewConnectionBackToReconnecting() + { + const string name = "stale-sub-connfail"; + var gate = new TaskCompletionSource(); + + var oldAdapter = HealthyAdapter("sub-old"); + oldAdapter.SubscribeAsync(Tag, Arg.Any(), Arg.Any()) + .Returns(_ => gate.Task); + var newAdapter = HealthyAdapter("sub-new"); + ExpectNewAdapter(newAdapter); + + var actor = CreateConnectionActor(name, oldAdapter); + AwaitCondition(() => HealthTransitions(name, ConnectionHealth.Connected) >= 1, + TimeSpan.FromSeconds(5), TimeSpan.FromMilliseconds(10)); + + var probe = CreateTestProbe(); + probe.Send(actor, new SubscribeTagsRequest("c1", "inst1", name, [Tag], DateTimeOffset.UtcNow)); + AwaitCondition(() => CalledWith(oldAdapter, nameof(IDataConnection.SubscribeAsync), Tag), + TimeSpan.FromSeconds(5)); + + actor.Tell(new ReconfigureConnectionCommand(name, "OpcUa", Config(NewEndpoint), null, 3)); + AwaitCondition(() => HealthTransitions(name, ConnectionHealth.Connected) >= 2, + TimeSpan.FromSeconds(5), TimeSpan.FromMilliseconds(10)); + + var disconnectsBefore = HealthTransitions(name, ConnectionHealth.Disconnected); + + // InvalidOperationException is what IsConnectionLevelFailure classifies as "transport + // down", i.e. the row that asks the state machine to reconnect. + gate.SetException(new InvalidOperationException("adapter not connected")); + + probe.FishForMessage( + m => m.CorrelationId == "c1" && !m.Success, TimeSpan.FromSeconds(5)); + + // Entering Reconnecting would push bad quality to every subscriber — the probe is one. + probe.ExpectNoMsg(TimeSpan.FromMilliseconds(500)); + Assert.Equal(disconnectsBefore, HealthTransitions(name, ConnectionHealth.Disconnected)); + Assert.Equal(2, HealthTransitions(name, ConnectionHealth.Connected)); + } + + // ── Alarm subscribe ── + + /// + /// The alarm twin. The stale completion lands first and (before the fix) stored its dead feed + /// id, so the LIVE feed's completion took the duplicate branch and released the feed the new + /// adapter had just established — a source that stays mirrored but never updates again. + /// + [Fact] + public void StaleAlarmSubscribeCompleted_DoesNotReleaseTheLiveAlarmFeed() + { + const string name = "stale-alarm-feed"; + var oldGate = new TaskCompletionSource(); + var newGate = new TaskCompletionSource(); + AlarmTransitionCallback? newCallback = null; + + var oldAdapter = HealthyAlarmAdapter("alarm-old"); + ((IAlarmSubscribableConnection)oldAdapter) + .SubscribeAlarmsAsync(Source, Arg.Any(), + Arg.Any(), Arg.Any()) + .Returns(_ => oldGate.Task); + + var newAdapter = HealthyAlarmAdapter("alarm-new"); + ((IAlarmSubscribableConnection)newAdapter) + .SubscribeAlarmsAsync(Source, Arg.Any(), + Arg.Do(c => newCallback = c), Arg.Any()) + .Returns(_ => newGate.Task); + ExpectNewAdapter(newAdapter); + + var actor = CreateConnectionActor(name, oldAdapter); + AwaitCondition(() => HealthTransitions(name, ConnectionHealth.Connected) >= 1, + TimeSpan.FromSeconds(5), TimeSpan.FromMilliseconds(10)); + + var probe = CreateTestProbe(); + probe.Send(actor, new SubscribeAlarmsRequest("c1", "inst1", name, Source, null, DateTimeOffset.UtcNow)); + AwaitCondition(() => CalledWith(oldAdapter, "SubscribeAlarmsAsync", Source), TimeSpan.FromSeconds(5)); + + actor.Tell(new ReconfigureConnectionCommand(name, "OpcUa", Config(NewEndpoint), null, 3)); + + // The subscriber registry survives the swap, so the reconnect re-subscribe re-opens the + // feed against the new adapter without the NativeAlarmActor doing anything. + AwaitCondition(() => CalledWith(newAdapter, "SubscribeAlarmsAsync", Source), + TimeSpan.FromSeconds(5), TimeSpan.FromMilliseconds(10)); + + // The dead feed answers first, the live one second — the ordering that made the live + // completion look redundant. + oldGate.SetResult("alarm-old"); + AwaitCondition(() => CalledWith(oldAdapter, "UnsubscribeAlarmsAsync", "alarm-old"), + TimeSpan.FromSeconds(5), TimeSpan.FromMilliseconds(10)); + newGate.SetResult("alarm-new"); + + AwaitCondition(() => newCallback != null, TimeSpan.FromSeconds(5)); + // Settle (and drain the swap's expected NativeAlarmSourceUnavailable) so the live + // completion has certainly been applied before asserting nothing released its feed. + probe.ReceiveWhile(TimeSpan.FromMilliseconds(300), m => m); + Assert.False(Called(newAdapter, "UnsubscribeAlarmsAsync")); + + // And the feed is genuinely live: a transition on it still routes to the subscriber. + newCallback!(Raise("Tank01.Hi", Source)); + probe.FishForMessage( + u => u.Transition.SourceObjectReference == Source, TimeSpan.FromSeconds(5)); + } +}