dohertj2/ScadaBridge
1
0
Fork 0
Code Issues 2 Pull Requests Actions Packages Projects Releases Wiki Activity

fix(dcl): generation-fence ConnectResult + roll back failed OPC UA subscribe batches (arch-review 2026-08-19 P2-3 + P2-4)

Browse Source 
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.
This commit is contained in:
Joseph Doherty
2026-08-19 14:40:56 -04:00
parent e2d149089c
commit 26ab6f8f09
5 changed files with 632 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Actors/DataConnectionActor.cs
+85 -6
View File
@@ -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);
}
/// <summary>
/// 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).
///
/// <para>
/// A success from the dead adapter used to drive <see cref="ReSubscribeAll"/> and
/// <see cref="BecomeConnected"/> 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 <c>AdapterDisconnected</c> 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.
/// </para>
///
/// <para>
/// Dropping cannot wedge the state machine: every site that bumps the generation
/// (<see cref="CountFailureAndMaybeFailover"/>, the failover branch of
/// <see cref="BecomeReconnecting"/>, and <see cref="HandleReconfigure"/> through it) arms an
/// <see cref="AttemptConnect"/> for the new generation, and the state stays Connecting /
/// Reconnecting until that attempt's own result arrives.
/// </para>
/// </summary>
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));
}
/// <summary>
/// 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 <see cref="Opc.Ua.ServiceResultException"/> 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.
///
/// <para>
/// Deliberately narrow. A per-node code — <c>BadNodeIdUnknown</c>, <c>BadNodeIdInvalid</c>,
/// <c>BadUserAccessDenied</c> — 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.
/// </para>
/// </summary>
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);
/// <summary>
/// Outcome of one <see cref="AttemptConnect"/>. 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 <see cref="IsStaleConnectResult"/>).
/// </summary>
internal record ConnectResult(bool Success, string? Error, int AdapterGeneration);
internal record AdapterDisconnected;
internal record TagValueReceived(string TagPath, TagValue Value, int AdapterGeneration);
internal record RetryTagResolution;
src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Adapters/RealOpcUaClient.cs
+111 -40
View File
@@ -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<DataShard>();
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);
}
/// <summary>
/// Places every resolved node on a data shard and issues ONE ApplyChanges per touched shard.
/// Callers must hold <see cref="_applyGate"/>. <paramref name="created"/> and
/// <paramref name="touched"/> are filled as the work proceeds so a throw can be unwound by
/// <see cref="UnwindCreatedItems"/>.
/// </summary>
private async Task PlaceAndApplyAsync(
List<(int Index, string NodeId, NodeId Resolved)> resolved,
Action<string, object?, DateTime, uint> onValueChanged,
List<(int Index, string NodeId, string Handle, MonitoredItem Item)> created,
List<DataShard> 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);
}
/// <summary>
/// Rolls back the monitored items a FAILED batch had already placed, so a throw leaves no
/// residue behind: no <c>_callbacks</c> entry (a leak that also keeps a dead callback
/// reachable from the SDK's publish thread), no <c>_itemShard</c> entry, no phantom shard
/// occupancy, and no item left on the shard's local item list. Callers must hold
/// <see cref="_applyGate"/>.
///
/// <para>
/// 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.
/// </para>
/// </summary>
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);
}
/// <summary>
/// Returns the shard this item belongs on per
/// <see cref="MonitoredItemShardPlanner"/> — first shard with free capacity, else a
tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Actors/DataConnectionOpcUaFaultClassificationTests.cs
+109
View File
@@ -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;
/// <summary>
/// Arch-review 2026-08-19 <b>P2-4</b> (classification half) — how a thrown
/// <see cref="ServiceResultException"/> out of a batch subscribe is classified.
///
/// <para>
/// <b>The gap these pin.</b> <c>IsConnectionLevelFailure</c> knew only about
/// <c>InvalidOperationException</c> / socket / timeout / IO faults, so the OPC UA SDK's own way of
/// reporting a dead session — a <c>ServiceResultException</c> 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.
/// </para>
///
/// <para>
/// The classification must stay narrow: a per-node status (<c>BadNodeIdUnknown</c>) 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.
/// </para>
/// </summary>
public class DataConnectionOpcUaFaultClassificationTests : TestKit
{
private readonly ISiteHealthCollector _health = Substitute.For<ISiteHealthCollector>();
private readonly IDataConnectionFactory _factory = Substitute.For<IDataConnectionFactory>();
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);
/// <summary>
/// A bad-session fault is a CONNECTION fault: the request fails and the actor enters
/// Reconnecting (which pushes bad quality for the connection).
///
/// <para><b>Fails before the fix:</b> the chunk was treated per-tag, so the request was acked
/// Success and the connection stayed nominally up on a dead session.</para>
/// </summary>
[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<SubscribeTagsResponse>(m => !m.Success && m.ErrorMessage != null, TimeSpan.FromSeconds(5));
ExpectMsg<ConnectionQualityChanged>(q => q.Quality == QualityCode.Bad, TimeSpan.FromSeconds(5));
}
/// <summary>
/// The narrowness control. <c>BadNodeIdUnknown</c> is a per-node fault: the tags go unresolved
/// (bad quality, retried on the backoff timer) and the connection is NOT torn down.
/// </summary>
[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<SubscribeTagsResponse>(m => m.Success, TimeSpan.FromSeconds(5));
Assert.Equal(0, HealthTransitions(name, ConnectionHealth.Disconnected));
}
}
tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Actors/DataConnectionStaleConnectResultTests.cs
+211
View File
@@ -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;
/// <summary>
/// Arch-review 2026-08-19 <b>P2-3</b> — the connect leg of the stale-completion family that
/// <c>DataConnectionStaleCompletionTests</c> pins for the subscribe leg.
///
/// <para>
/// <b>The gap these pin.</b> <c>ConnectResult</c> 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
/// <c>ReSubscribeAll</c> and <c>BecomeConnected</c> 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
/// <c>AdapterDisconnected</c> 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.
/// </para>
///
/// <para>
/// 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.
/// </para>
/// </summary>
[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<ISiteHealthCollector>();
private readonly IDataConnectionFactory _factory = Substitute.For<IDataConnectionFactory>();
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<string, string> 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<IDictionary<string, string>>(d => d["Endpoint"] == endpoint))
.Returns(adapter);
/// <summary>An adapter whose connect completes when <paramref name="connect"/> does.</summary>
private static IDataConnection Adapter(Task connect, string subscriptionId)
{
var adapter = Substitute.For<IDataConnection>();
adapter.ConnectAsync(Arg.Any<IDictionary<string, string>>(), Arg.Any<CancellationToken>())
.Returns(connect);
adapter.Status.Returns(ConnectionHealth.Connected);
adapter.SubscribeAsync(Arg.Any<string>(), Arg.Any<SubscriptionCallback>(), Arg.Any<CancellationToken>())
.Returns(subscriptionId);
adapter.ReadAsync(Arg.Any<string>(), Arg.Any<CancellationToken>())
.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);
/// <summary>
/// Drives the actor to: connected on the first adapter with one live tag subscription, then
/// reconfigured onto <paramref name="mid"/> (whose connect is still in flight), then
/// reconfigured again onto <paramref name="last"/> (whose connect is also still in flight).
/// The mid adapter's pending connect is now two generations stale.
/// </summary>
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<SubscribeTagsResponse>(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;
}
/// <summary>
/// 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.
///
/// <para><b>Fails before the fix:</b> the actor entered Connected and re-subscribed
/// <c>Tag</c> on the never-connected last adapter.</para>
/// </summary>
[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)));
}
/// <summary>
/// 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
/// <c>ConnectAsync</c> was launched against an adapter already connecting.
///
/// <para><b>Fails before the fix:</b> the last adapter was asked to connect a second time
/// one <c>ReconnectInterval</c> after the stale failure landed.</para>
/// </summary>
[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));
}
/// <summary>
/// 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.
/// </summary>
[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));
}
}
tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Adapters/RealOpcUaClientBatchRollbackTests.cs
+116
View File
@@ -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;
/// <summary>
/// Arch-review 2026-08-19 <b>P2-4</b> (rollback half) — a batch subscribe whose
/// <c>ApplyChanges</c> throws must leave NO residue behind.
///
/// <para>
/// <b>The gap this pins.</b> <c>CreateSubscriptionsCoreAsync</c> registers each monitored item as
/// it places it (<c>_callbacks</c>, <c>_itemShard</c>, the shard's item count and the SDK
/// subscription's own item list) and only issues <c>ApplyChanges</c> 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.
/// </para>
///
/// <para>
/// The seam is reached with a data shard whose SDK <see cref="Subscription"/> was never created,
/// which is precisely how the SDK reports "you have no usable subscription": placement and
/// <c>AddItem</c> succeed, <c>ApplyChangesAsync</c> throws. Private state is read by reflection
/// rather than widened for the test.
/// </para>
/// </summary>
[Trait("Category", "Unit")]
public class RealOpcUaClientBatchRollbackTests
{
private static object Field(object target, string name) =>
target.GetType().GetField(name, BindingFlags.Instance | BindingFlags.NonPublic)!.GetValue(target)!;
/// <summary>
/// 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.
/// </summary>
private static Subscription ArrangeUncreatedShard(RealOpcUaClient client)
{
var session = Substitute.For<ISession>();
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<string, Action<string, object?, DateTime, uint>>)
Field(client, "_callbacks"));
Assert.Empty((ConcurrentDictionary<string, Subscription>)Field(client, "_itemShard"));
Assert.Empty((ConcurrentDictionary<string, MonitoredItem>)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<string, Action<string, object?, DateTime, uint>>)
Field(client, "_callbacks"));
Assert.Empty((ConcurrentDictionary<string, Subscription>)Field(client, "_itemShard"));
Assert.Equal(0, ShardItemCount(client));
}
}