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

fix(dcl): P2-5 seed probe-resolved tags; P2-6 close the Connecting/Reconnecting AdapterDisconnected gap

Browse Source 
P2-5 — the tag-resolution PROBE was the only subscribe path that did not seed.
HandleSubscribe carries seeds back on SubscribeCompleted and ReSubscribeAll re-seeds
after a reconnect, because a fresh advise only yields a value on the source's next
CHANGE. HandleRetryTagResolution did neither, so a tag that resolved LATE (device still
booting, item created after deployment) and then never changed had no value path at all.
OPC UA masks it with the initial data-change notification on subscription creation; an
MxGateway advise can legitimately return VT_EMPTY, so a static tag stayed Uncertain for
the life of the connection. New SeedNewlyResolvedTags reads the round's newly resolved
tags and delivers them through the normal generation-guarded TagValueReceived path —
directly rather than on the completion message (the probe only covers tags already
registered in _instancesByTag, so the initial subscribe's ordering hazard does not
apply), and as its own fire-and-forget task so the seed reads never delay the
BatchSubscribeCompleted that re-arms the retry timer.

P2-6 — only the Connected state handled AdapterDisconnected; in Connecting and
Reconnecting it fell through to Unhandled. A connect that succeeds and drops again before
its piped ConnectResult is applied therefore delivered the disconnect FIRST, where it was
discarded, and ConnectResult(true) then declared the connection up on a dead adapter. An
adapter raises Disconnected at most once per successful connect (OpcUaDataConnection's
once-only _disconnectFired, reset only by the next successful ConnectAsync), so a
connection with no subscriptions — nothing to fail at connection level and drive the
reconnect machine — sat Connected on a dead adapter until the process restarted. Both
states now record the disconnect in _disconnectDuringConnectAttempt (cleared at the top of
HandleAttemptConnect, so it only ever refers to the attempt in flight) and both result
handlers demote such a "success" to the ordinary connect-FAILURE tail — counting toward
the failover budget and re-arming the retry timer. A flag rather than Stash: the stash has
no selective drop, so a stashed disconnect would be replayed after the next successful
connect and cost a spurious reconnect cycle.

AdapterDisconnected is now stamped with the adapter generation (hooked via a
per-adapter closure instead of a method group; SwapAdapter hooks AFTER the bump), and
IsStaleDisconnect drops a notification from a swapped-out adapter in all three states —
the disconnect twin of IsStaleConnectResult. SwapAdapter's unhook cannot stop an event
already executing on a background thread, so without the fence an endpoint the operator
has just reconfigured AWAY from could knock the healthy replacement into a reconnect.

Tests: DataConnectionProbeSeedTests + DataConnectionDisconnectRaceTests (4 new). The
race is reproduced by ORDERING, not timing — the connect is held on a gate, Disconnected
is raised while the actor sits idle, and a GetHealthReport round trip is the mailbox
barrier; no sleeps. All 4 were proved to fail against the pre-fix behaviour via a
targeted revert of the four behaviour sites (the stale-generation test cannot be proved
by a plain revert — the record carried no generation before this change, so the test
would not compile; noted in its doc comment). Full DCL suite 346/346 green.
This commit is contained in:
Joseph Doherty
2026-08-19 16:24:30 -04:00
parent 65b3c23239
commit 6823e7e3fa
3 changed files with 612 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Actors/DataConnectionActor.cs
+264 -11
View File
@@ -234,6 +234,51 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers
/// </summary>
private IActorRef _self = null!;
/// <summary>
/// The delegate currently hooked to <c>_adapter.Disconnected</c>. It closes over the
/// <see cref="_adapterGeneration"/> in force when it was hooked, so the resulting
/// <see cref="AdapterDisconnected"/> carries the generation of the adapter that raised it —
/// a method group could not. Kept in a field so <see cref="SwapAdapter"/> and
/// <see cref="PostStop"/> unhook exactly the instance they hooked.
/// </summary>
private Action? _disconnectedHandler;
/// <summary>
/// Arch-review 2026-08-19 <b>P2-6</b>. Set when an <see cref="AdapterDisconnected"/> for the
/// CURRENT adapter generation is received while the actor is still Connecting/Reconnecting —
/// i.e. the adapter connected and dropped again before the piped
/// <see cref="ConnectResult"/> for that attempt reached the mailbox.
///
/// <para>
/// Neither state used to handle the message at all, so it fell through to
/// <c>Unhandled</c> and the subsequent <c>ConnectResult(true)</c> declared the connection up
/// on an adapter that was already dead. Nothing recovers from there: an adapter fires
/// <c>Disconnected</c> at most once per successful connect (<c>OpcUaDataConnection</c>'s
/// once-only <c>_disconnectFired</c> guard, reset only by the next successful
/// <c>ConnectAsync</c>), so a connection with zero subscriptions — nothing to fail at
/// connection level and drive the reconnect machine — sat Connected on a dead adapter until
/// the process restarted.
/// </para>
///
/// <para>
/// It is cleared at the top of <see cref="HandleAttemptConnect"/>, so it only ever refers to
/// the attempt currently in flight: a disconnect belonging to a session that ended BEFORE
/// this attempt began (e.g. the adapter's <c>Disconnected</c> trailing a connection-level
/// subscribe failure that already drove the actor into Reconnecting) is discarded rather than
/// replayed against the fresh session. That is also why the flag is used instead of
/// <c>Stash</c>: the stash has no selective drop, so a stashed disconnect would be replayed
/// after the next successful connect and cost a spurious reconnect cycle.
/// </para>
///
/// <para>
/// Residual (accepted, and in the safe direction): a disconnect raised by the PREVIOUS
/// session but delivered after the next attempt has already cleared the flag is attributed to
/// the new attempt, costing one extra reconnect cycle. Being wedged Connected on a dead
/// adapter is the failure this trades against.
/// </para>
/// </summary>
private bool _disconnectDuringConnectAttempt;
/// <summary>
/// Initializes the data connection actor with its adapter and configuration.
/// </summary>
@@ -283,23 +328,43 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers
_self = Self;
// Listen for unexpected adapter disconnections
_adapter.Disconnected += OnAdapterDisconnected;
HookAdapterDisconnected();
BecomeConnecting();
}
private void OnAdapterDisconnected()
/// <summary>
/// Subscribes to the CURRENT adapter's <c>Disconnected</c> event with a delegate that closes
/// over the generation in force right now, so the marshalled
/// <see cref="AdapterDisconnected"/> identifies the adapter that raised it. Callers must
/// have bumped <see cref="_adapterGeneration"/> for the new adapter first
/// (see <see cref="SwapAdapter"/>).
/// </summary>
private void HookAdapterDisconnected()
{
// Marshal the event onto the actor's message loop using captured _self reference.
// This runs on a background thread (gRPC stream reader), so Self would throw.
_self.Tell(new AdapterDisconnected());
var generation = _adapterGeneration;
var self = _self;
// Marshal the event onto the actor's message loop using the captured _self reference.
// This runs on a background thread (keep-alive / gRPC stream reader), so Self would throw.
_disconnectedHandler = () => self.Tell(new AdapterDisconnected(generation));
_adapter.Disconnected += _disconnectedHandler;
}
/// <summary>Unhooks the delegate installed by <see cref="HookAdapterDisconnected"/>, if any.</summary>
private void UnhookAdapterDisconnected()
{
if (_disconnectedHandler is null)
return;
_adapter.Disconnected -= _disconnectedHandler;
_disconnectedHandler = null;
}
/// <inheritdoc />
protected override void PostStop()
{
_log.Info("DataConnectionActor [{0}] stopping — disposing adapter", _connectionName);
_adapter.Disconnected -= OnAdapterDisconnected;
UnhookAdapterDisconnected();
// S9: fire-and-forget, but observe a faulted dispose so a failing adapter
// teardown is logged rather than swallowed into an unobserved task exception.
_adapter.DisposeAsync().AsTask().ContinueWith(
@@ -340,6 +405,14 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers
case ConnectResult result:
HandleConnectResult(result);
break;
case AdapterDisconnected disconnected:
// P2-6 — the adapter connected and dropped again before this attempt's
// ConnectResult was applied. Remember it (see _disconnectDuringConnectAttempt)
// rather than acting now: acting would put the actor into Reconnecting and the
// in-flight ConnectResult(true) would then be applied THERE, re-subscribing and
// declaring the same dead adapter Connected anyway.
NotePendingDisconnect(disconnected);
break;
case SubscribeTagsRequest:
case WriteTagRequest:
case WriteTagBatchRequest:
@@ -467,8 +540,13 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers
case QualityFlushTick:
FlushQualityCountersIfDirty();
break;
case AdapterDisconnected:
HandleDisconnect();
case AdapterDisconnected disconnected:
// P2-6 — fence the disconnect on its adapter generation. The outgoing adapter's
// event can fire concurrently with SwapAdapter's unhook, so a Disconnected from
// the endpoint we have just failed over (or reconfigured) AWAY from could arrive
// here and knock the healthy replacement back into a reconnect cycle.
if (!IsStaleDisconnect(disconnected))
HandleDisconnect();
break;
case ReconfigureConnectionCommand rc:
HandleReconfigure(rc);
@@ -600,14 +678,17 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers
/// </summary>
private void SwapAdapter(IDictionary<string, string> newConfig)
{
_adapter.Disconnected -= OnAdapterDisconnected;
UnhookAdapterDisconnected();
_ = _adapter.DisposeAsync().AsTask();
_adapter = _factory.Create(_protocolType, newConfig);
_connectionDetails = newConfig;
_adapter.Disconnected += OnAdapterDisconnected;
_adapterGeneration++;
// Hooked AFTER the bump so the new adapter's handler stamps the NEW generation; the
// outgoing adapter's handler was captured with the old one and is now recognised as
// stale by IsStaleDisconnect even if it fires concurrently with the unhook above.
HookAdapterDisconnected();
}
/// <summary>
@@ -702,6 +783,12 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers
case ConnectResult result:
HandleReconnectResult(result);
break;
case AdapterDisconnected disconnected:
// P2-6 — same gap as Connecting: a reconnect that succeeds and drops again before
// its ConnectResult is applied would otherwise be declared Connected on a dead
// adapter that will never raise Disconnected a second time.
NotePendingDisconnect(disconnected);
break;
case SubscribeTagsRequest:
case WriteTagRequest:
case WriteTagBatchRequest:
@@ -777,6 +864,10 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers
private void HandleAttemptConnect()
{
_log.Debug("[{0}] Attempting connection...", _connectionName);
// P2-6 — scope the pending-disconnect flag to THIS attempt. Anything noted before now
// belongs to a session that is already over and must not be replayed against the one
// about to start.
_disconnectDuringConnectAttempt = false;
var self = Self;
// Stamp the generation this attempt belongs to, exactly as every other piped
// completion does. A failover or an in-place reconfigure landing while ConnectAsync
@@ -827,6 +918,75 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers
return true;
}
/// <summary>
/// True when this <see cref="AdapterDisconnected"/> was raised by an adapter that has since
/// been swapped out (failover or in-place reconfigure), in which case it is dropped — the
/// disconnect twin of <see cref="IsStaleConnectResult"/> (arch-review 2026-08-19 P2-6).
/// <see cref="SwapAdapter"/> unhooks the outgoing adapter, but the event can already be
/// executing on a background thread when it does, so the message can still be delivered
/// after the swap.
/// </summary>
private bool IsStaleDisconnect(AdapterDisconnected disconnected)
{
if (disconnected.AdapterGeneration == _adapterGeneration)
return false;
_log.Debug(
"[{0}] Dropping disconnect notification from adapter generation {1} (current {2}) — "
+ "the adapter that raised it has been replaced",
_connectionName, disconnected.AdapterGeneration, _adapterGeneration);
return true;
}
/// <summary>
/// Records a current-generation disconnect that arrived while a connect attempt was still in
/// flight, so <see cref="HandleConnectResult"/> / <see cref="HandleReconnectResult"/> can
/// refuse to declare the connection up. See <see cref="_disconnectDuringConnectAttempt"/>.
/// </summary>
private void NotePendingDisconnect(AdapterDisconnected disconnected)
{
if (IsStaleDisconnect(disconnected))
return;
_log.Warning(
"[{0}] Adapter reported a disconnect while a connect attempt was still in flight — "
+ "the attempt's result will be treated as a failed connection", _connectionName);
_disconnectDuringConnectAttempt = true;
}
/// <summary>
/// Reads and clears <see cref="_disconnectDuringConnectAttempt"/>, logging the demotion. True
/// means the successful connect result being applied belongs to a session the adapter has
/// already reported as dropped.
/// </summary>
private bool ConsumeDisconnectDuringConnectAttempt()
{
if (!_disconnectDuringConnectAttempt)
return false;
_disconnectDuringConnectAttempt = false;
_log.Warning(
"[{0}] Connect succeeded but the adapter had already reported the session lost — "
+ "treating the attempt as a failed connect", _connectionName);
return true;
}
/// <summary>
/// Applies a demoted connect success (see <see cref="ConsumeDisconnectDuringConnectAttempt"/>)
/// exactly as a connect FAILURE: the session the "success" describes is already gone, so it
/// must count toward the failover budget and re-arm the retry timer rather than declare the
/// connection up. Verbatim the tail of both result handlers' failure branches, which is the
/// point — a flapping endpoint that drops inside every connect window still reaches its
/// backup instead of retrying the same dead primary forever.
/// </summary>
private void DemoteConnectAttemptToFailure()
{
_consecutiveFailures++;
CountFailureAndMaybeFailover(
"adapter reported the session lost before the connect result was applied");
Timers.StartSingleTimer("reconnect", new AttemptConnect(), _options.ReconnectInterval);
}
private void HandleConnectResult(ConnectResult result)
{
if (IsStaleConnectResult(result))
@@ -834,6 +994,16 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers
if (result.Success)
{
// P2-6 — a disconnect that raced ahead of this result means the session this
// "success" describes is already gone; going Connected here would strand the
// connection on a dead adapter that never raises Disconnected again. Stay in
// Connecting and retry, exactly as a failed connect would.
if (ConsumeDisconnectDuringConnectAttempt())
{
DemoteConnectAttemptToFailure();
return;
}
_log.Info("[{0}] Connection established", _connectionName);
BecomeConnected();
}
@@ -858,6 +1028,15 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers
if (result.Success)
{
// P2-6 — same demotion as the initial connect: the reconnect landed but the adapter
// has already reported the new session lost, so re-subscribing and going Connected
// would wedge on a dead adapter. Stay in Reconnecting and retry.
if (ConsumeDisconnectDuringConnectAttempt())
{
DemoteConnectAttemptToFailure();
return;
}
_log.Info("[{0}] Reconnected successfully on {1} endpoint", _connectionName, _activeEndpoint);
_consecutiveFailures = 0;
@@ -2131,10 +2310,79 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers
}));
}
// NOTE: `self` is the Self reference captured on the actor thread above — this runs on
// a background task, where Akka's Self property is invalid.
SeedNewlyResolvedTags(self, adapter, generation, rows);
return new BatchSubscribeCompleted(rows, generation, BatchSubscribeSource.ResolutionProbe);
}).PipeTo(self);
}
/// <summary>
/// Reads the current value of every tag this probe round has just resolved and delivers it
/// through the normal generation-guarded <see cref="TagValueReceived"/> path — the same seed
/// the initial subscribe (<see cref="HandleSubscribe"/>) and the reconnect re-subscribe
/// (<see cref="ReSubscribeAll"/>) already perform, which the resolution probe was missing
/// (arch-review 2026-08-19 <b>P2-5</b>).
///
/// <para>
/// A newly advised subscription only starts producing values on the source's next CHANGE.
/// OPC UA masks that with an initial data-change notification on subscription creation, but
/// an MxGateway advise can legitimately return <c>VT_EMPTY</c>, so a tag that resolved LATE
/// (device still booting, item created after deployment) and then never changes had no value
/// path at all — its attribute stayed Uncertain for the life of the connection, healed only
/// by an unrelated reconnect.
/// </para>
///
/// <para>
/// Unlike <see cref="HandleSubscribe"/>, the seeds are delivered directly rather than carried
/// back on the completion message: the probe only ever covers tags already present in
/// <see cref="_subscriptionsByInstance"/> / <see cref="_instancesByTag"/>, so the
/// registration-ordering hazard that forces the initial subscribe to defer does not exist
/// here. It runs as its own fire-and-forget task (mirroring the reconnect re-seed) so the
/// seed reads — bounded by <c>SeedOverallTimeout</c>, not by the subscribe — never delay the
/// <see cref="BatchSubscribeCompleted"/> that carries the round's resolution bookkeeping and
/// re-arms the retry timer. Values that arrive after a failover or a drop to Reconnecting are
/// discarded by the generation guard / state handler exactly like any other tag value.
/// </para>
/// </summary>
/// <param name="self">
/// The actor's own reference, captured on the actor thread by the caller — this method is
/// invoked from a background task, where Akka's <c>Self</c> property is invalid.
/// </param>
/// <param name="adapter">The adapter the probe round ran against.</param>
/// <param name="generation">The adapter generation the probe round was stamped with.</param>
/// <param name="rows">The probe round's per-tag subscribe results.</param>
private void SeedNewlyResolvedTags(
IActorRef self, IDataConnection adapter, int generation, IReadOnlyList<SubscribeTagResult> rows)
{
var resolved = new List<string>(rows.Count);
foreach (var row in rows)
{
if (row is { Success: true, SubscriptionId: not null })
resolved.Add(row.TagPath);
}
if (resolved.Count == 0)
return;
_ = Task.Run(async () =>
{
try
{
var seeds = await SeedTagsAsync(adapter, resolved);
foreach (var seed in seeds)
self.Tell(new TagValueReceived(seed.TagPath, seed.Value, generation));
}
catch (Exception ex)
{
// Fire-and-forget: guarantee a trace rather than an unobserved task fault.
// SeedTagsAsync already swallows per-read errors, so reaching here is unexpected.
_log.Warning("[{0}] Tag-resolution re-seed task faulted: {1}", _connectionName, ex.Message);
}
});
}
/// <summary>
/// Arms the single-shot tag-resolution retry timer at the current backoff interval
/// when unresolved tags remain, and cancels it when none do. Single-shot (rescheduled
@@ -2961,7 +3209,12 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers
/// rather than applied to its replacement (see <see cref="IsStaleConnectResult"/>).
/// </summary>
internal record ConnectResult(bool Success, string? Error, int AdapterGeneration);
internal record AdapterDisconnected;
/// <summary>
/// The adapter's <c>Disconnected</c> event marshalled onto the actor thread, stamped with the
/// <see cref="_adapterGeneration"/> the raising adapter was hooked at so a notification from a
/// swapped-out adapter can be dropped (<see cref="IsStaleDisconnect"/>).
/// </summary>
internal record AdapterDisconnected(int AdapterGeneration);
internal record TagValueReceived(string TagPath, TagValue Value, int AdapterGeneration);
internal record RetryTagResolution;
tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Actors/DataConnectionDisconnectRaceTests.cs
+242
View File
@@ -0,0 +1,242 @@
using Akka.Actor;
using Akka.TestKit.Xunit2;
using NSubstitute;
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Protocol;
using ZB.MOM.WW.ScadaBridge.Commons.Messages.DataConnection;
using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums;
using ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Actors;
using ZB.MOM.WW.ScadaBridge.HealthMonitoring;
namespace ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests.Actors;
/// <summary>
/// Arch-review 2026-08-19 <b>P2-6</b> — the <c>AdapterDisconnected</c> gap in the
/// Connecting/Reconnecting states, and the generation fence that keeps a swapped-out adapter's
/// disconnect from disturbing its replacement.
///
/// <para>
/// <b>The gap these pin.</b> Only the Connected state handled <c>AdapterDisconnected</c>; in
/// Connecting and Reconnecting it fell through to <c>Unhandled</c>. A connect that succeeds and
/// drops again before its piped <c>ConnectResult</c> is applied therefore delivered the disconnect
/// FIRST, where it was discarded, and the <c>ConnectResult(true)</c> that followed declared the
/// connection up on an adapter that was already dead. Nothing recovers from there: an adapter
/// raises <c>Disconnected</c> at most once per successful connect (<c>OpcUaDataConnection</c>'s
/// once-only <c>_disconnectFired</c> guard, reset only by the next successful <c>ConnectAsync</c>),
/// so on a connection with no subscriptions — nothing to fail at connection level and drive the
/// reconnect machine — the actor sat Connected on a dead adapter for the life of the process.
/// </para>
///
/// <para>
/// The fix must not weaken the Medium-wave generation fencing: a disconnect raised by an adapter
/// that has since been swapped out (failover, or the row-97 in-place reconfigure) must still be
/// dropped, or the endpoint the operator has just corrected AWAY from could knock the healthy
/// replacement back into a reconnect cycle.
/// </para>
///
/// <para>
/// <b>Determinism note.</b> The race is reproduced by ORDERING, not by timing: the adapter's
/// connect is held on a gate, the <c>Disconnected</c> event is raised while the actor sits idle in
/// Connecting, and a <c>GetHealthReport</c> round trip is used as a mailbox barrier proving the
/// disconnect has been processed before the connect result is released. No sleeps.
/// </para>
/// </summary>
[Collection("DataConnectionManagerActor")]
public class DataConnectionDisconnectRaceTests : TestKit
{
private const string FirstEndpoint = "opc.tcp://first-plc:4840";
private const string SecondEndpoint = "opc.tcp://second-plc:4840";
private readonly ISiteHealthCollector _health = Substitute.For<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 static IDataConnection Adapter(Task connect)
{
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("sub-1");
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);
/// <summary>
/// Mailbox barrier: the reply to a message sent AFTER <paramref name="actor"/> was told
/// something proves the earlier message has already been processed (Akka mailboxes are FIFO
/// per sender pair, and every state of this actor answers <c>GetHealthReport</c>). Used
/// instead of a settle delay so the ordering the defect depends on is asserted, not waited on.
/// </summary>
private void Barrier(IActorRef actor)
{
var probe = CreateTestProbe();
probe.Send(actor, new DataConnectionActor.GetHealthReport());
probe.ExpectMsg<DataConnectionHealthReport>(TimeSpan.FromSeconds(5));
}
/// <summary>
/// The headline case: the adapter connects and drops again before the actor has applied the
/// connect result. The actor must NOT settle in Connected — the adapter will never raise
/// <c>Disconnected</c> again, so Connected would be terminal — and must retry instead.
///
/// <para><b>Fails before the fix:</b> the disconnect was Unhandled in Connecting, the actor
/// went straight to Connected and never reconnected — exactly one <c>ConnectAsync</c> for the
/// life of the actor.</para>
/// </summary>
[Fact]
public void DisconnectBeforeConnectResult_RetriesInsteadOfGoingConnected()
{
const string name = "disconnect-before-connect-result";
var firstConnect = new TaskCompletionSource();
// Attempt #2 onward never completes, so "the actor did not go Connected on the dead
// session" is a stable assertion rather than a race against the next retry tick.
var laterConnects = new TaskCompletionSource();
var connects = 0;
var adapter = Substitute.For<IDataConnection>();
adapter.ConnectAsync(Arg.Any<IDictionary<string, string>>(), Arg.Any<CancellationToken>())
.Returns(_ => Interlocked.Increment(ref connects) == 1 ? firstConnect.Task : laterConnects.Task);
adapter.Status.Returns(ConnectionHealth.Connected);
adapter.ReadAsync(Arg.Any<string>(), Arg.Any<CancellationToken>())
.Returns(new ReadResult(false, null, null));
var actor = CreateConnectionActor(name, adapter);
// The actor is in Connecting with ConnectAsync in flight on the gate.
AwaitCondition(() => Volatile.Read(ref connects) >= 1,
TimeSpan.FromSeconds(5), TimeSpan.FromMilliseconds(10));
// The link drops. This is the real ordering: the adapter's keep-alive thread raises
// Disconnected while the connect's piped result is still on its way.
adapter.Disconnected += Raise.Event<Action>();
Barrier(actor);
// Only now does the (already obsolete) success land.
firstConnect.SetResult();
// The retry cycle runs, rather than the actor parking Connected on a dead adapter.
AwaitCondition(() => Volatile.Read(ref connects) >= 2,
TimeSpan.FromSeconds(5), TimeSpan.FromMilliseconds(10));
Assert.Equal(0, HealthTransitions(name, ConnectionHealth.Connected));
}
/// <summary>
/// The Reconnecting twin of the case above — the same gap existed there, and a reconnect is
/// where a flapping endpoint spends most of its time.
///
/// <para><b>Fails before the fix:</b> the disconnect was Unhandled in Reconnecting and the
/// stale success re-subscribed and declared the connection up on the dead adapter.</para>
/// </summary>
[Fact]
public void DisconnectBeforeReconnectResult_StaysReconnecting()
{
const string name = "disconnect-before-reconnect-result";
var firstConnect = new TaskCompletionSource();
var secondConnect = new TaskCompletionSource();
// Attempt #3 onward never completes, so the actor parks in Reconnecting and the
// "did not go Connected on the dead session" assertion is stable rather than a race
// against the next retry tick.
var laterConnects = new TaskCompletionSource();
var connects = 0;
var adapter = Substitute.For<IDataConnection>();
adapter.ConnectAsync(Arg.Any<IDictionary<string, string>>(), Arg.Any<CancellationToken>())
.Returns(_ => Interlocked.Increment(ref connects) switch
{
1 => firstConnect.Task,
2 => secondConnect.Task,
_ => laterConnects.Task
});
adapter.Status.Returns(ConnectionHealth.Connected);
adapter.ReadAsync(Arg.Any<string>(), Arg.Any<CancellationToken>())
.Returns(new ReadResult(false, null, null));
var actor = CreateConnectionActor(name, adapter);
firstConnect.SetResult();
AwaitCondition(() => HealthTransitions(name, ConnectionHealth.Connected) >= 1,
TimeSpan.FromSeconds(5), TimeSpan.FromMilliseconds(10));
// Drop while Connected → Reconnecting, and the retry timer launches attempt #2.
adapter.Disconnected += Raise.Event<Action>();
AwaitCondition(() => Volatile.Read(ref connects) >= 2,
TimeSpan.FromSeconds(5), TimeSpan.FromMilliseconds(10));
// The reconnect drops again before its result is applied.
adapter.Disconnected += Raise.Event<Action>();
Barrier(actor);
secondConnect.SetResult();
// The demoted success re-arms the retry timer instead of declaring the link up again.
AwaitCondition(() => Volatile.Read(ref connects) >= 3,
TimeSpan.FromSeconds(5), TimeSpan.FromMilliseconds(10));
Assert.Equal(1, HealthTransitions(name, ConnectionHealth.Connected));
}
/// <summary>
/// The fencing guard rail: a disconnect stamped with a superseded adapter generation must
/// still be dropped. <c>SwapAdapter</c> unhooks the outgoing adapter, but the event can
/// already be running on a background thread when it does, so the message can outlive the
/// swap — and acting on it would take a healthy, freshly reconfigured connection down.
///
/// <para>
/// <b>Proof of failure.</b> This one cannot be shown against the pre-fix tree by stashing:
/// <c>AdapterDisconnected</c> carried no generation at all before this change, so the test
/// would not compile. It was proved by a targeted revert instead — restoring
/// <c>case AdapterDisconnected: HandleDisconnect(); break;</c> in the Connected state while
/// keeping the stamped record — which makes the assertion below fail with 2 Disconnected
/// transitions instead of 1.
/// </para>
/// </summary>
[Fact]
public void StaleGenerationDisconnect_IsIgnored()
{
const string name = "stale-generation-disconnect";
var first = Adapter(Task.CompletedTask);
var second = Adapter(Task.CompletedTask);
_factory.Create("OpcUa", Arg.Is<IDictionary<string, string>>(d => d["Endpoint"] == SecondEndpoint))
.Returns(second);
var actor = CreateConnectionActor(name, first);
AwaitCondition(() => HealthTransitions(name, ConnectionHealth.Connected) >= 1,
TimeSpan.FromSeconds(5), TimeSpan.FromMilliseconds(10));
// In-place reconfigure: swaps the adapter and bumps the generation 0 → 1. The single
// Disconnected transition it publishes on the way through Reconnecting is the baseline.
actor.Tell(new ReconfigureConnectionCommand(name, "OpcUa", Config(SecondEndpoint), null, 3));
AwaitCondition(() => HealthTransitions(name, ConnectionHealth.Connected) >= 2,
TimeSpan.FromSeconds(5), TimeSpan.FromMilliseconds(10));
Assert.Equal(1, HealthTransitions(name, ConnectionHealth.Disconnected));
// The endpoint we have just moved away from finally reports its loss.
actor.Tell(new DataConnectionActor.AdapterDisconnected(0));
Barrier(actor);
Assert.Equal(1, HealthTransitions(name, ConnectionHealth.Disconnected));
Assert.Equal(2, HealthTransitions(name, ConnectionHealth.Connected));
}
}
tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Actors/DataConnectionProbeSeedTests.cs
+106
View File
@@ -0,0 +1,106 @@
using Akka.Actor;
using Akka.TestKit.Xunit2;
using NSubstitute;
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Protocol;
using ZB.MOM.WW.ScadaBridge.Commons.Messages.DataConnection;
using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums;
using ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Actors;
using ZB.MOM.WW.ScadaBridge.HealthMonitoring;
namespace ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests.Actors;
/// <summary>
/// Arch-review 2026-08-19 <b>P2-5</b> — the tag-resolution PROBE path did not seed the tags it
/// resolved.
///
/// <para>
/// <b>The gap this pins.</b> Both other subscribe paths read each newly advised tag's current
/// value and deliver it (<c>HandleSubscribe</c> carries the seeds back on <c>SubscribeCompleted</c>;
/// <c>ReSubscribeAll</c> re-seeds after a reconnect), because a fresh advise only produces a value
/// on the source's next CHANGE. <c>HandleRetryTagResolution</c> did not. OPC UA masks that with an
/// initial data-change notification on subscription creation, but an MxGateway advise can
/// legitimately return <c>VT_EMPTY</c> — so a STATIC tag that resolved LATE (device still booting,
/// item created after deployment) had no value path at all and its attribute stayed Uncertain for
/// the life of the connection.
/// </para>
/// </summary>
[Collection("DataConnectionManagerActor")]
public class DataConnectionProbeSeedTests : TestKit
{
private const string Tag = "plant/static/setpoint";
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(150),
WriteTimeout = TimeSpan.FromSeconds(5),
SeedReadMaxAttempts = 1,
SeedReadRetryDelay = TimeSpan.FromMilliseconds(10)
};
private IActorRef CreateConnectionActor(string name, IDataConnection adapter) =>
Sys.ActorOf(Props.Create(() =>
new DataConnectionActor(name, adapter, _options, _health, _factory, "OpcUa")), name);
private bool HasEnteredConnectedState(string connectionName) =>
_health.ReceivedCalls().Any(c =>
c.GetMethodInfo().Name == nameof(ISiteHealthCollector.UpdateConnectionHealth) &&
c.GetArguments() is [string name, ConnectionHealth.Connected, ..] &&
name == connectionName);
/// <summary>
/// A tag that fails to resolve on the initial subscribe, resolves on the first probe round,
/// and then never changes again must still reach its subscriber with a Good value.
///
/// <para>
/// The adapter here models exactly the MxGateway shape the finding calls out: the advise
/// succeeds on the retry but the subscription callback is NEVER invoked (a <c>VT_EMPTY</c>
/// advise on a static item), so the seed read is the tag's only possible value source.
/// </para>
///
/// <para><b>Fails before the fix:</b> the probe resolved the tag and stopped there — the
/// subscriber received only the Bad-quality resolution-failure push from the initial
/// subscribe, and this fish for a Good value timed out.</para>
/// </summary>
[Fact]
public void ProbeResolvedTag_IsSeeded_SoAStaticTagLeavesUncertain()
{
const string name = "probe-seed";
var subscribeCalls = 0;
var adapter = Substitute.For<IDataConnection>();
adapter.ConnectAsync(Arg.Any<IDictionary<string, string>>(), Arg.Any<CancellationToken>())
.Returns(Task.CompletedTask);
adapter.Status.Returns(ConnectionHealth.Connected);
adapter.SubscribeAsync(Tag, Arg.Any<SubscriptionCallback>(), Arg.Any<CancellationToken>())
.Returns(_ => Interlocked.Increment(ref subscribeCalls) == 1
// Initial subscribe: the device is still booting — a genuine (non
// connection-level) resolution failure, so the tag lands in _unresolvedTags.
? Task.FromException<string>(new KeyNotFoundException("node not found yet"))
// Probe round: the advise succeeds. The callback is deliberately never invoked.
: Task.FromResult("sub-1"));
adapter.ReadAsync(Tag, Arg.Any<CancellationToken>())
.Returns(new ReadResult(true, new TagValue(42, QualityCode.Good, DateTimeOffset.UtcNow), null));
var actor = CreateConnectionActor(name, adapter);
AwaitCondition(() => HasEnteredConnectedState(name),
TimeSpan.FromSeconds(5), TimeSpan.FromMilliseconds(10));
var subscriber = CreateTestProbe();
subscriber.Send(actor, new SubscribeTagsRequest("c1", "inst1", name, [Tag], DateTimeOffset.UtcNow));
// The initial subscribe fails: Bad-quality push, then the ack.
subscriber.FishForMessage<SubscribeTagsResponse>(_ => true, TimeSpan.FromSeconds(5));
// The probe round resolves the tag; the seed read is the only thing that can now
// deliver a value, because the advise callback never fires for this static item.
var seeded = subscriber.FishForMessage<TagValueUpdate>(
m => m.TagPath == Tag && m.Quality == QualityCode.Good, TimeSpan.FromSeconds(10));
Assert.Equal(42, seeded.Value);
Assert.True(Volatile.Read(ref subscribeCalls) >= 2, "the probe round must have re-subscribed the tag");
}
}