From 26ab6f8f09f7691b5daf432afb6c80bdcf97ae0f Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Wed, 19 Aug 2026 14:40:56 -0400 Subject: [PATCH] fix(dcl): generation-fence ConnectResult + roll back failed OPC UA subscribe batches (arch-review 2026-08-19 P2-3 + P2-4) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit P2-3. ConnectResult was the last piped completion with no adapter generation on it, so a connect launched against generation N and answered after the adapter had been swapped (failover, or the row-97 in-place reconfigure) was applied to its REPLACEMENT. A stale success drove ReSubscribeAll + BecomeConnected against an adapter that had never connected, and the replacement's own result then landed in the Connected state where nothing handles it. With a bad new endpoint the batch subscribes failed at connection level with a log line only, no AdapterDisconnected ever fired, and the idempotent-reconfigure guard refused a re-send of the same configuration — only a restart recovered the connection. A stale FAILURE was as wrong the other way: it counted toward failover and re-armed the reconnect timer underneath the attempt genuinely in flight. HandleAttemptConnect now stamps the generation and both handlers open with the same guard the subscribe leg got in c4d3c8b8, dropping a mismatched result whole. The adapter instance is deliberately NOT carried: unlike a subscribe completion a connect result holds no adapter-scoped handles to release, and SwapAdapter has already disposed the outgoing adapter. Dropping cannot wedge the machine — every generation bump (CountFailureAndMaybeFailover, BecomeReconnecting's failover branch, HandleReconfigure through it) arms an AttemptConnect for the new generation, and the state stays Connecting/Reconnecting until that answers. P2-4. CreateSubscriptionsCoreAsync registers each monitored item as it places it (_callbacks, _itemShard, shard budget, the SDK subscription's item list) and only then issues ApplyChanges, so a throw there propagated with every registration still in place and nothing holding a handle to release them. The actor reads the throw as a whole-chunk fault and re-issues the SAME batch on its backoff timer, so each probe grew the bookkeeping again. Placement + apply moved to PlaceAndApplyAsync behind a try/catch that unwinds via the existing ReleaseFailedItem (per-item try, so one failure neither stops the rest nor replaces the original fault); no ApplyChanges is re-issued, and items a partial apply did reach the server with are reclaimed when the shard and session are torn down on reconnect. That reconnect is the other half: IsConnectionLevelFailure knew nothing about ServiceResultException, which is how the SDK reports a dead session — so a dead session classified as a per-tag resolution failure for the whole chunk. It now also matches a narrow set of session/channel status codes (BadSessionIdInvalid, BadSessionClosed, BadSessionNotActivated, BadSecureChannelClosed, BadSecureChannelIdInvalid, BadServerNotConnected, BadConnectionClosed, BadCommunicationError, BadNotConnected). Per-node codes such as BadNodeIdUnknown stay per-tag — that is exactly the device-still-booting case the resolution timer exists for. Tests: DataConnectionStaleConnectResultTests (stale success neither connects nor re-subscribes against the new adapter; stale failure does not disturb the current attempt; the current generation's result is still processed after a stale drop) — all 3 fail with the guards removed. DataConnectionOpcUaFaultClassificationTests (bad-session drives reconnect — fails unfixed; bad-node-id stays per-tag). RealOpcUaClientBatchRollbackTests (no residual registrations after an ApplyChanges throw; repeated failed batches do not grow the bookkeeping) — both fail with the rollback removed. Full DCL suite 342/342 green. --- .../Actors/DataConnectionActor.cs | 91 +++++++- .../Adapters/RealOpcUaClient.cs | 151 +++++++++---- ...ConnectionOpcUaFaultClassificationTests.cs | 109 +++++++++ .../DataConnectionStaleConnectResultTests.cs | 211 ++++++++++++++++++ .../RealOpcUaClientBatchRollbackTests.cs | 116 ++++++++++ 5 files changed, 632 insertions(+), 46 deletions(-) create mode 100644 tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Actors/DataConnectionOpcUaFaultClassificationTests.cs create mode 100644 tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Actors/DataConnectionStaleConnectResultTests.cs create mode 100644 tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Adapters/RealOpcUaClientBatchRollbackTests.cs diff --git a/src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Actors/DataConnectionActor.cs b/src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Actors/DataConnectionActor.cs index c5bb81d0..c5a16c11 100644 --- a/src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Actors/DataConnectionActor.cs +++ b/src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Actors/DataConnectionActor.cs @@ -778,16 +778,60 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers { _log.Debug("[{0}] Attempting connection...", _connectionName); 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 + // is still in flight swaps the adapter and bumps the counter, and the OUTGOING + // adapter's answer must never be applied to its replacement. + var generation = _adapterGeneration; _adapter.ConnectAsync(_connectionDetails).ContinueWith(t => { if (t.IsCompletedSuccessfully) - return new ConnectResult(true, null); - return new ConnectResult(false, t.Exception?.GetBaseException().Message); + return new ConnectResult(true, null, generation); + return new ConnectResult(false, t.Exception?.GetBaseException().Message, generation); }).PipeTo(self); } + /// + /// True when this connect result was produced by an adapter that has since been swapped + /// out, in which case it is dropped WHOLE (arch-review 2026-08-19 P2-3). + /// + /// + /// A success from the dead adapter used to drive and + /// against the NEW, never-connected adapter — and the new + /// adapter's own result then arrived in the Connected state, where nothing handles it. If the + /// new endpoint was bad the batch subscribes failed at connection level with a log line only, + /// no AdapterDisconnected ever fired, and the idempotent-reconfigure guard rejected + /// re-sending the same configuration: only a restart recovered the connection. A failure from + /// the dead adapter is just as wrong — it would count toward failover and re-arm the reconnect + /// timer underneath the attempt that is genuinely in flight. + /// + /// + /// + /// Dropping cannot wedge the state machine: every site that bumps the generation + /// (, the failover branch of + /// , and through it) arms an + /// for the new generation, and the state stays Connecting / + /// Reconnecting until that attempt's own result arrives. + /// + /// + private bool IsStaleConnectResult(ConnectResult result) + { + if (result.AdapterGeneration == _adapterGeneration) + return false; + + _log.Debug( + "[{0}] Dropping {1} connect result from adapter generation {2} (current {3}) — " + + "the adapter it ran against has been replaced", + _connectionName, result.Success ? "successful" : "failed", + result.AdapterGeneration, _adapterGeneration); + return true; + } + private void HandleConnectResult(ConnectResult result) { + if (IsStaleConnectResult(result)) + return; + if (result.Success) { _log.Info("[{0}] Connection established", _connectionName); @@ -809,6 +853,9 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers private void HandleReconnectResult(ConnectResult result) { + if (IsStaleConnectResult(result)) + return; + if (result.Success) { _log.Info("[{0}] Reconnected successfully on {1} endpoint", _connectionName, _activeEndpoint); @@ -1538,11 +1585,38 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers { var baseEx = ex is AggregateException agg ? agg.GetBaseException() : ex; return baseEx is InvalidOperationException - or System.Net.Sockets.SocketException - or TimeoutException - or System.IO.IOException; + or System.Net.Sockets.SocketException + or TimeoutException + or System.IO.IOException + || (baseEx is Opc.Ua.ServiceResultException sre && IsConnectionLevelStatus(sre.StatusCode)); } + /// + /// OPC UA status codes that mean the SESSION or the SECURE CHANNEL is gone, not that one + /// node is unusable (arch-review 2026-08-19 P2-4). The SDK reports a dead session by + /// throwing out of the batch subscribe, which + /// classified as a per-tag fault: every tag in the chunk was marked unresolved and each + /// backoff probe re-issued the whole batch against the same dead session, growing the + /// adapter's monitored-item bookkeeping on every round instead of reconnecting once. + /// + /// + /// Deliberately narrow. A per-node code — BadNodeIdUnknown, BadNodeIdInvalid, + /// BadUserAccessDenied — must stay a tag-resolution failure, because that is exactly + /// the "device still booting / tag path wrong" case the retry timer exists for; treating it + /// as connection-level would put the connection into a reconnect loop over one bad tag path. + /// + /// + private static bool IsConnectionLevelStatus(uint statusCode) => statusCode is + Opc.Ua.StatusCodes.BadSessionIdInvalid + or Opc.Ua.StatusCodes.BadSessionClosed + or Opc.Ua.StatusCodes.BadSessionNotActivated + or Opc.Ua.StatusCodes.BadSecureChannelClosed + or Opc.Ua.StatusCodes.BadSecureChannelIdInvalid + or Opc.Ua.StatusCodes.BadServerNotConnected + or Opc.Ua.StatusCodes.BadConnectionClosed + or Opc.Ua.StatusCodes.BadCommunicationError + or Opc.Ua.StatusCodes.BadNotConnected; + private void HandleUnsubscribe(UnsubscribeTagsRequest request) { _log.Debug("[{0}] Unsubscribing all tags for instance {1}", @@ -2881,7 +2955,12 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers // ── Internal messages ── internal record AttemptConnect; - internal record ConnectResult(bool Success, string? Error); + /// + /// Outcome of one . Carries the adapter generation it ran + /// against so a result produced by an adapter that has since been swapped out is dropped + /// rather than applied to its replacement (see ). + /// + internal record ConnectResult(bool Success, string? Error, int AdapterGeneration); internal record AdapterDisconnected; internal record TagValueReceived(string TagPath, TagValue Value, int AdapterGeneration); internal record RetryTagResolution; diff --git a/src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Adapters/RealOpcUaClient.cs b/src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Adapters/RealOpcUaClient.cs index 26dd75bf..663440e2 100644 --- a/src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Adapters/RealOpcUaClient.cs +++ b/src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Adapters/RealOpcUaClient.cs @@ -621,48 +621,21 @@ public class RealOpcUaClient : IOpcUaClient var created = new List<(int Index, string NodeId, string Handle, MonitoredItem Item)>(resolved.Count); var touched = new List(); - foreach (var (index, nodeId, startNode) in resolved) + try { - var shard = await PlaceOnDataShardAsync(cancellationToken); - var handle = Guid.NewGuid().ToString(); - var monitoredItem = new MonitoredItem(shard.Subscription.DefaultItem) - { - DisplayName = nodeId, - StartNodeId = startNode, - AttributeId = Attributes.Value, - SamplingInterval = _options.SamplingIntervalMs, - QueueSize = (uint)_options.QueueSize, - DiscardOldest = _options.DiscardOldest, - Filter = BuildDataChangeFilter(_options.Deadband) - }; - - _callbacks[handle] = onValueChanged; - monitoredItem.Notification += (_, e) => - { - if (e.NotificationValue is MonitoredItemNotification notification) - { - var value = notification.Value?.Value; - var timestamp = notification.Value?.SourceTimestamp ?? DateTime.UtcNow; - var statusCode = notification.Value?.StatusCode.Code ?? 0; - - if (_callbacks.TryGetValue(handle, out var cb)) - { - cb(nodeId, value, timestamp, statusCode); - } - } - }; - - shard.Subscription.AddItem(monitoredItem); - shard.ItemCount++; - _itemShard[handle] = shard.Subscription; - created.Add((index, nodeId, handle, monitoredItem)); - if (!touched.Contains(shard)) - touched.Add(shard); + await PlaceAndApplyAsync(resolved, onValueChanged, created, touched, cancellationToken); + } + catch + { + // Anything thrown from placement or ApplyChanges (a dead session is the common + // one) leaves every item created so far registered in _callbacks / _itemShard and + // counted against its shard's budget, with nothing to ever release them: the + // actor sees the throw as a whole-chunk fault and re-issues the SAME batch on its + // backoff timer, so each probe grew the bookkeeping again (arch-review 2026-08-19 + // P2-4). Unwind before propagating. + UnwindCreatedItems(created); + throw; } - - // ONE ApplyChanges per touched shard — the whole point of the batch seam. - foreach (var shard in touched) - await shard.Subscription.ApplyChangesAsync(cancellationToken); foreach (var (index, nodeId, handle, item) in created) { @@ -687,6 +660,104 @@ public class RealOpcUaClient : IOpcUaClient }, cancellationToken); } + /// + /// Places every resolved node on a data shard and issues ONE ApplyChanges per touched shard. + /// Callers must hold . and + /// are filled as the work proceeds so a throw can be unwound by + /// . + /// + private async Task PlaceAndApplyAsync( + List<(int Index, string NodeId, NodeId Resolved)> resolved, + Action onValueChanged, + List<(int Index, string NodeId, string Handle, MonitoredItem Item)> created, + List touched, + CancellationToken cancellationToken) + { + foreach (var (index, nodeId, startNode) in resolved) + { + var shard = await PlaceOnDataShardAsync(cancellationToken); + var handle = Guid.NewGuid().ToString(); + var monitoredItem = new MonitoredItem(shard.Subscription.DefaultItem) + { + DisplayName = nodeId, + StartNodeId = startNode, + AttributeId = Attributes.Value, + SamplingInterval = _options.SamplingIntervalMs, + QueueSize = (uint)_options.QueueSize, + DiscardOldest = _options.DiscardOldest, + Filter = BuildDataChangeFilter(_options.Deadband) + }; + + _callbacks[handle] = onValueChanged; + monitoredItem.Notification += (_, e) => + { + if (e.NotificationValue is MonitoredItemNotification notification) + { + var value = notification.Value?.Value; + var timestamp = notification.Value?.SourceTimestamp ?? DateTime.UtcNow; + var statusCode = notification.Value?.StatusCode.Code ?? 0; + + if (_callbacks.TryGetValue(handle, out var cb)) + { + cb(nodeId, value, timestamp, statusCode); + } + } + }; + + shard.Subscription.AddItem(monitoredItem); + shard.ItemCount++; + _itemShard[handle] = shard.Subscription; + created.Add((index, nodeId, handle, monitoredItem)); + if (!touched.Contains(shard)) + touched.Add(shard); + } + + // ONE ApplyChanges per touched shard — the whole point of the batch seam. + foreach (var shard in touched) + await shard.Subscription.ApplyChangesAsync(cancellationToken); + } + + /// + /// Rolls back the monitored items a FAILED batch had already placed, so a throw leaves no + /// residue behind: no _callbacks entry (a leak that also keeps a dead callback + /// reachable from the SDK's publish thread), no _itemShard entry, no phantom shard + /// occupancy, and no item left on the shard's local item list. Callers must hold + /// . + /// + /// + /// Best-effort and exception-safe by design: it runs on the failure path, so one item that + /// cannot be released must not stop the rest from being released and must not replace the + /// original fault. No ApplyChanges is re-issued — the session that just threw is generally + /// dead, and where a partial apply DID reach the server the items it created are reclaimed + /// when the shard and session are torn down on the reconnect that the connection-level + /// classification of that fault now drives. + /// + /// + private void UnwindCreatedItems( + List<(int Index, string NodeId, string Handle, MonitoredItem Item)> created) + { + if (created.Count == 0) + return; + + foreach (var (_, nodeId, handle, item) in created) + { + try + { + ReleaseFailedItem(handle, item); + } + catch (Exception ex) + { + _logger.LogWarning(ex, + "OPC UA batch subscribe rollback could not release the monitored item for {NodeId}", + nodeId); + } + } + + _logger.LogWarning( + "OPC UA batch subscribe failed — rolled back {Count} partially created monitored item(s)", + created.Count); + } + /// /// Returns the shard this item belongs on per /// — first shard with free capacity, else a diff --git a/tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Actors/DataConnectionOpcUaFaultClassificationTests.cs b/tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Actors/DataConnectionOpcUaFaultClassificationTests.cs new file mode 100644 index 00000000..40ad6104 --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Actors/DataConnectionOpcUaFaultClassificationTests.cs @@ -0,0 +1,109 @@ +using Akka.Actor; +using Akka.TestKit.Xunit2; +using NSubstitute; +using Opc.Ua; +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-4 (classification half) — how a thrown +/// out of a batch subscribe is classified. +/// +/// +/// The gap these pin. IsConnectionLevelFailure knew only about +/// InvalidOperationException / socket / timeout / IO faults, so the OPC UA SDK's own way of +/// reporting a dead session — a ServiceResultException carrying a bad-session status — +/// was treated as a per-tag resolution failure for the WHOLE chunk. Every tag went unresolved and +/// the backoff probe re-issued the same batch against the same dead session forever, growing the +/// adapter's monitored-item bookkeeping each round, instead of reconnecting once. +/// +/// +/// +/// The classification must stay narrow: a per-node status (BadNodeIdUnknown) is exactly the +/// "device still booting / wrong tag path" case the resolution timer exists for, and must NOT put +/// the connection into a reconnect loop. +/// +/// +public class DataConnectionOpcUaFaultClassificationTests : TestKit +{ + private readonly ISiteHealthCollector _health = Substitute.For(); + private readonly IDataConnectionFactory _factory = Substitute.For(); + + private static DataConnectionOptions Options() => new() + { + ReconnectInterval = TimeSpan.FromMilliseconds(100), + TagResolutionRetryInterval = TimeSpan.FromMilliseconds(150), + TagResolutionRetryMaxInterval = TimeSpan.FromMilliseconds(600), + WriteTimeout = TimeSpan.FromSeconds(5), + SeedReadMaxAttempts = 1, + SeedReadRetryDelay = TimeSpan.FromMilliseconds(10), + SubscribeBatchSize = 5, + SubscribeBatchDelay = TimeSpan.FromMilliseconds(50), + SeedReadBatchSize = 4, + SeedReadMaxParallelism = 2, + SeedOverallTimeout = TimeSpan.FromSeconds(30), + QualityFlushInterval = TimeSpan.FromMilliseconds(200) + }; + + private IActorRef CreateActor(FakeBatchDataConnection adapter, string name) => + Sys.ActorOf(Props.Create(() => new DataConnectionActor( + name, adapter, Options(), _health, _factory, "OpcUa")), name); + + 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); + + /// + /// A bad-session fault is a CONNECTION fault: the request fails and the actor enters + /// Reconnecting (which pushes bad quality for the connection). + /// + /// Fails before the fix: the chunk was treated per-tag, so the request was acked + /// Success and the connection stayed nominally up on a dead session. + /// + [Fact] + public void BadSessionServiceResultException_IsConnectionLevel_AndDrivesReconnect() + { + const string name = "opcua-bad-session"; + var adapter = new FakeBatchDataConnection + { + BatchSubscribeThrows = () => + new ServiceResultException(StatusCodes.BadSessionIdInvalid, "session is no longer valid") + }; + var actor = CreateActor(adapter, name); + + actor.Tell(new SubscribeTagsRequest("c1", "inst1", name, ["tag1", "tag2"], DateTimeOffset.UtcNow)); + + ExpectMsg(m => !m.Success && m.ErrorMessage != null, TimeSpan.FromSeconds(5)); + ExpectMsg(q => q.Quality == QualityCode.Bad, TimeSpan.FromSeconds(5)); + } + + /// + /// The narrowness control. BadNodeIdUnknown is a per-node fault: the tags go unresolved + /// (bad quality, retried on the backoff timer) and the connection is NOT torn down. + /// + [Fact] + public void BadNodeIdServiceResultException_StaysPerTag_AndDoesNotTearDownTheConnection() + { + const string name = "opcua-bad-nodeid"; + var adapter = new FakeBatchDataConnection + { + BatchSubscribeThrows = () => + new ServiceResultException(StatusCodes.BadNodeIdUnknown, "no such node") + }; + var actor = CreateActor(adapter, name); + + actor.Tell(new SubscribeTagsRequest("c1", "inst1", name, ["tag1", "tag2"], DateTimeOffset.UtcNow)); + + // Per-tag treatment: bad quality per tag, and the request itself is still acked Success + // (the tags are registered and retried, exactly as an unresolved tag path is). + FishForMessage(m => m.Success, TimeSpan.FromSeconds(5)); + Assert.Equal(0, HealthTransitions(name, ConnectionHealth.Disconnected)); + } +} diff --git a/tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Actors/DataConnectionStaleConnectResultTests.cs b/tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Actors/DataConnectionStaleConnectResultTests.cs new file mode 100644 index 00000000..d525c243 --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Actors/DataConnectionStaleConnectResultTests.cs @@ -0,0 +1,211 @@ +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-3 — the connect leg of the stale-completion family that +/// DataConnectionStaleCompletionTests pins for the subscribe leg. +/// +/// +/// The gap these pin. ConnectResult was piped untagged, so a connect launched +/// against adapter generation N and answered after the adapter had been swapped (failover, or the +/// row-97 in-place reconfigure) was applied to its REPLACEMENT. A stale success drove +/// ReSubscribeAll and BecomeConnected against an adapter that had never connected; +/// the replacement's own result then landed in the Connected state, where nothing handles it. With +/// a bad new endpoint the batch subscribes failed at connection level with a log line only, no +/// AdapterDisconnected ever fired, and the idempotent-reconfigure guard refused a re-send +/// of the same configuration — only a restart recovered the connection. A stale FAILURE was just +/// as wrong: it counted toward failover and re-armed the reconnect timer underneath the attempt +/// that was genuinely in flight. +/// +/// +/// +/// The state machine must also stay able to finish the CURRENT attempt after a stale result was +/// dropped — Connecting/Reconnecting is held until the matching-generation result arrives. +/// +/// +[Collection("DataConnectionManagerActor")] +public class DataConnectionStaleConnectResultTests : TestKit +{ + private const string FirstEndpoint = "opc.tcp://first-plc:4840"; + private const string MidEndpoint = "opc.tcp://mid-plc:4840"; + private const string LastEndpoint = "opc.tcp://last-plc:4840"; + private const string Tag = "sensor/temp"; + + 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 void ExpectAdapterFor(string endpoint, IDataConnection adapter) => + _factory.Create("OpcUa", Arg.Is>(d => d["Endpoint"] == endpoint)) + .Returns(adapter); + + /// An adapter whose connect completes when does. + private static IDataConnection Adapter(Task connect, string subscriptionId) + { + 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(subscriptionId); + 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); + + private static int CallCount(IDataConnection adapter, string method) => + adapter.ReceivedCalls().Count(c => c.GetMethodInfo().Name == method); + + /// + /// Drives the actor to: connected on the first adapter with one live tag subscription, then + /// reconfigured onto (whose connect is still in flight), then + /// reconfigured again onto (whose connect is also still in flight). + /// The mid adapter's pending connect is now two generations stale. + /// + private IActorRef ArrangeTwoGenerationsBehind( + string name, IDataConnection first, IDataConnection mid, IDataConnection last) + { + ExpectAdapterFor(MidEndpoint, mid); + ExpectAdapterFor(LastEndpoint, last); + + var actor = CreateConnectionActor(name, first); + 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)); + probe.FishForMessage(m => m.Success, TimeSpan.FromSeconds(5)); + + actor.Tell(new ReconfigureConnectionCommand(name, "OpcUa", Config(MidEndpoint), null, 3)); + AwaitCondition(() => CallCount(mid, nameof(IDataConnection.ConnectAsync)) >= 1, + TimeSpan.FromSeconds(5), TimeSpan.FromMilliseconds(10)); + + actor.Tell(new ReconfigureConnectionCommand(name, "OpcUa", Config(LastEndpoint), null, 3)); + AwaitCondition(() => CallCount(last, nameof(IDataConnection.ConnectAsync)) >= 1, + TimeSpan.FromSeconds(5), TimeSpan.FromMilliseconds(10)); + + return actor; + } + + /// + /// The headline case: a success from the adapter the operator has just corrected AWAY from + /// must not declare the connection up, and must not re-issue the subscription registry + /// against the replacement adapter that has not connected yet. + /// + /// Fails before the fix: the actor entered Connected and re-subscribed + /// Tag on the never-connected last adapter. + /// + [Fact] + public void StaleConnectSuccess_DoesNotConnect_AndDoesNotResubscribeAgainstTheNewAdapter() + { + const string name = "stale-connect-success"; + var midGate = new TaskCompletionSource(); + var lastGate = new TaskCompletionSource(); + + var first = Adapter(Task.CompletedTask, "sub-first"); + var mid = Adapter(midGate.Task, "sub-mid"); + var last = Adapter(lastGate.Task, "sub-last"); + + ArrangeTwoGenerationsBehind(name, first, mid, last); + + // The two-generations-old connect finally answers — successfully. + midGate.SetResult(); + // Settle: nothing here subscribes the default test actor, so this is purely a wait long + // enough for the piped result to have been received and dropped. + ExpectNoMsg(TimeSpan.FromMilliseconds(500)); + + Assert.Equal(1, HealthTransitions(name, ConnectionHealth.Connected)); + Assert.Equal(0, CallCount(last, nameof(IDataConnection.SubscribeAsync))); + } + + /// + /// A stale FAILURE must not touch the attempt that is genuinely in flight. Acting on it + /// counted toward the failover budget and re-armed the reconnect timer, so a second + /// ConnectAsync was launched against an adapter already connecting. + /// + /// Fails before the fix: the last adapter was asked to connect a second time + /// one ReconnectInterval after the stale failure landed. + /// + [Fact] + public void StaleConnectFailure_DoesNotDisturbTheCurrentAttempt() + { + const string name = "stale-connect-failure"; + var midGate = new TaskCompletionSource(); + var lastGate = new TaskCompletionSource(); + + var first = Adapter(Task.CompletedTask, "sub-first"); + var mid = Adapter(midGate.Task, "sub-mid"); + var last = Adapter(lastGate.Task, "sub-last"); + + ArrangeTwoGenerationsBehind(name, first, mid, last); + + midGate.SetException(new Exception("connection refused")); + // Comfortably longer than ReconnectInterval, so a re-armed timer would have fired. + ExpectNoMsg(TimeSpan.FromMilliseconds(600)); + + Assert.Equal(1, CallCount(last, nameof(IDataConnection.ConnectAsync))); + Assert.Equal(1, HealthTransitions(name, ConnectionHealth.Connected)); + } + + /// + /// Dropping a stale result must not wedge the state machine: the actor stays in Reconnecting + /// and the CURRENT generation's own result is still processed normally — connection up, and + /// the whole subscription registry transparently re-subscribed against the live adapter. + /// + [Fact] + public void CurrentGenerationConnectResult_IsStillProcessed_AfterAStaleOneWasDropped() + { + const string name = "stale-connect-then-live"; + var midGate = new TaskCompletionSource(); + var lastGate = new TaskCompletionSource(); + + var first = Adapter(Task.CompletedTask, "sub-first"); + var mid = Adapter(midGate.Task, "sub-mid"); + var last = Adapter(lastGate.Task, "sub-last"); + + ArrangeTwoGenerationsBehind(name, first, mid, last); + + midGate.SetResult(); + ExpectNoMsg(TimeSpan.FromMilliseconds(300)); + Assert.Equal(1, HealthTransitions(name, ConnectionHealth.Connected)); + + // Now the live adapter answers. + lastGate.SetResult(); + + AwaitCondition(() => HealthTransitions(name, ConnectionHealth.Connected) >= 2, + TimeSpan.FromSeconds(5), TimeSpan.FromMilliseconds(10)); + AwaitCondition(() => CallCount(last, nameof(IDataConnection.SubscribeAsync)) >= 1, + TimeSpan.FromSeconds(5), TimeSpan.FromMilliseconds(10)); + } +} diff --git a/tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Adapters/RealOpcUaClientBatchRollbackTests.cs b/tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Adapters/RealOpcUaClientBatchRollbackTests.cs new file mode 100644 index 00000000..4d6ea521 --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Adapters/RealOpcUaClientBatchRollbackTests.cs @@ -0,0 +1,116 @@ +using System.Collections; +using System.Collections.Concurrent; +using System.Reflection; +using NSubstitute; +using Opc.Ua; +using Opc.Ua.Client; +using ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Adapters; + +namespace ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests.Adapters; + +/// +/// Arch-review 2026-08-19 P2-4 (rollback half) — a batch subscribe whose +/// ApplyChanges throws must leave NO residue behind. +/// +/// +/// The gap this pins. CreateSubscriptionsCoreAsync registers each monitored item as +/// it places it (_callbacks, _itemShard, the shard's item count and the SDK +/// subscription's own item list) and only issues ApplyChanges once the whole chunk is +/// placed. A throw there — a dead session being the common cause — propagated with every one of +/// those registrations still in place and nothing left holding a handle to release them. The actor +/// sees the throw as a whole-chunk fault and re-issues the SAME batch on its backoff timer, so +/// each probe grew the bookkeeping again: unbounded shard growth and phantom shard occupancy. +/// +/// +/// +/// The seam is reached with a data shard whose SDK was never created, +/// which is precisely how the SDK reports "you have no usable subscription": placement and +/// AddItem succeed, ApplyChangesAsync throws. Private state is read by reflection +/// rather than widened for the test. +/// +/// +[Trait("Category", "Unit")] +public class RealOpcUaClientBatchRollbackTests +{ + private static object Field(object target, string name) => + target.GetType().GetField(name, BindingFlags.Instance | BindingFlags.NonPublic)!.GetValue(target)!; + + /// + /// Gives the client a mocked session plus ONE pre-placed (never created) data shard, so a + /// batch subscribe places its items and then throws out of ApplyChanges. + /// + private static Subscription ArrangeUncreatedShard(RealOpcUaClient client) + { + var session = Substitute.For(); + session.NamespaceUris.Returns(new NamespaceTable()); + typeof(RealOpcUaClient) + .GetField("_session", BindingFlags.Instance | BindingFlags.NonPublic)! + .SetValue(client, session); + + // The parameterless ctor is obsolete in the current SDK in favour of a telemetry-context + // overload; nothing here publishes telemetry, and the alternative would be to hand-build a + // context purely to satisfy the signature. +#pragma warning disable CS0618 + var subscription = new Subscription { DisplayName = "uncreated-shard" }; +#pragma warning restore CS0618 + + var shardType = typeof(RealOpcUaClient).GetNestedType("DataShard", BindingFlags.NonPublic)!; + var shard = Activator.CreateInstance( + shardType, + BindingFlags.Instance | BindingFlags.Public | BindingFlags.NonPublic, + null, [subscription], null)!; + ((IList)Field(client, "_dataShards")).Add(shard); + + return subscription; + } + + private static int ShardItemCount(RealOpcUaClient client) + { + var shard = ((IList)Field(client, "_dataShards"))[0]!; + return (int)shard.GetType().GetProperty("ItemCount")!.GetValue(shard)!; + } + + [Fact] + public async Task ApplyChangesThrow_RollsBackEveryPartiallyCreatedItem() + { + var client = new RealOpcUaClient(); + var subscription = ArrangeUncreatedShard(client); + + var fault = await Record.ExceptionAsync(() => client.CreateSubscriptionsAsync( + ["i=2258", "i=2259", "i=2260"], + (_, _, _, _) => { })); + + // Positive control: the batch really did fail at ApplyChanges, not before placement. + Assert.NotNull(fault); + + // Nothing registered survives the throw — pre-call state, exactly. + Assert.Empty((ConcurrentDictionary>) + Field(client, "_callbacks")); + Assert.Empty((ConcurrentDictionary)Field(client, "_itemShard")); + Assert.Empty((ConcurrentDictionary)Field(client, "_monitoredItems")); + Assert.Equal(0, ShardItemCount(client)); + Assert.Empty(subscription.MonitoredItems); + } + + [Fact] + public async Task RepeatedFailedBatches_DoNotGrowTheAdapterBookkeeping() + { + // The actual operational damage: the actor re-issues the same chunk on its backoff timer, + // so an un-rolled-back batch compounded on every probe. + var client = new RealOpcUaClient(); + ArrangeUncreatedShard(client); + + for (var attempt = 0; attempt < 3; attempt++) + { + var fault = await Record.ExceptionAsync(() => client.CreateSubscriptionsAsync( + ["i=2258", "i=2259"], + (_, _, _, _) => { })); + Assert.NotNull(fault); + } + + Assert.Empty((ConcurrentDictionary>) + Field(client, "_callbacks")); + Assert.Empty((ConcurrentDictionary)Field(client, "_itemShard")); + Assert.Equal(0, ShardItemCount(client)); + } +}