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

feat(dcl+siteruntime): reconfigure a LIVE data connection in place — register row 97

Browse Source 
`DataConnectionManagerActor.HandleCreateConnection` is create-only (warns and
returns for a name it already holds), so `EnsureDclConnection`'s documented
"a changed config re-issues the create so the DCL adopts it" was never true on
ANY node: a corrected endpoint/credential/backup/failover-count took effect only
when the connection was next built from scratch (node restart, singleton
failover, remove-then-re-add), while the DM's hash cache recorded it as applied.
Rows 89/90 made that honest with a Warning; this closes it.

New `ReconfigureConnectionCommand` (Commons, same shape as the create):

- `DeploymentManagerActor.EnsureDclConnection` now has three outcomes off its one
  per-name hash slot — unchanged -> nothing at all, new -> create, CHANGED ->
  reconfigure — and logs "Reconfigured DCL connection {Connection}
  (protocol={Protocol}) in place" at Information in place of the old Warning. The
  winner-resolution / arbitration / tombstone logic is untouched; only what
  happens after a change is detected changed.
- `DataConnectionManagerActor` FORWARDS the command to the owning connection
  actor, falling back to a create for a name it does not hold, so the command is
  safe whatever the manager's state.
- `DataConnectionActor` handles it in ALL THREE states and never stashes it — a
  connection pointed at a wrong or dead endpoint lives in Connecting/Reconnecting,
  which is exactly when a correction arrives; stashed it would wait on the broken
  configuration forever. Handling replaces protocol/primary/backup/failover-count,
  re-arms endpoint selection to Primary (the new config's primary is the intent,
  whatever the old config had failed over to), and calls the new shared
  `SwapAdapter` helper — now used by both failover sites too, so the swap exists
  once: unhook + fire-and-forget dispose -> _factory.Create -> rewire -> bump
  _adapterGeneration LAST. `BecomeReconnecting(afterReconfigure: true)` then
  reconnects immediately (StartSingleTimer replaces any armed reconnect, so no
  racing attempts) with the unstable-disconnect accounting and failover check
  suppressed — the link was dropped on purpose, and counting it as an unstable
  cycle could trip a failover onto the NEW backup before the new primary had been
  tried once. ReSubscribeAll/ReSubscribeAllAlarms then re-issue everything from
  the untouched subscription registries. Idempotent: a reconfigure carrying the
  configuration already in force is a no-op.

Stop-and-recreate was rejected as destructive: subscriptions live inside
DataConnectionActor and InstanceActor cancels its subscribe-retry timer on first
success, so a rebuilt actor would have no subscribers and nothing would ever
re-subscribe.

Tests: new DataConnectionReconfigureTests (7; 5 fail / 2 pass against a
behavioural revert). DclConnectionArbitrationTests gains two row-97 tests and its
CDC-arrival test now asserts the second push is a reconfigure; its ping-pong
guard also asserts zero reconfigures, since a flip would now show up there.
SiteRuntime-010 renamed to ..._ReissuesReconfigureCommand. 3 fail against a
reverted EnsureDclConnection. DataConnectionLayer.Tests 327/327,
SiteRuntime.Tests 643/643, solution build 0 warnings.

Docs: Component-DataConnectionLayer.md gains an "In-place reconfiguration"
section under Endpoint Redundancy; Component-SiteRuntime.md, and the
docs/components/ mirrors, updated; the arbitration decision memo gains an
appended §9 postscript resolving its residual 1 (residuals 2 and 3 stay open).
This commit is contained in:
Joseph Doherty
2026-08-17 06:19:13 -04:00
parent dcfcf822af
commit 7c1268d85c
12 changed files with 943 additions and 126 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Actors/DataConnectionActor.cs
+196 -78
View File
@@ -37,7 +37,12 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers
private readonly DataConnectionOptions _options;
private readonly ISiteHealthCollector _healthCollector;
private readonly IDataConnectionFactory _factory;
private readonly string _protocolType;
/// <summary>
/// Protocol of the configuration currently in force. NOT readonly: an in-place
/// reconfiguration (<see cref="ReconfigureConnectionCommand"/>) may change the protocol, and
/// the factory keys off it when the replacement adapter is built.
/// </summary>
private string _protocolType;
private readonly ISiteEventLogger? _siteEventLogger;
/// <summary>Stash for holding messages while the connection is not in the Connected state.</summary>
@@ -202,9 +207,13 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers
private readonly HashSet<string> _alarmSourcesWithoutSeparator = new(StringComparer.Ordinal);
private IDictionary<string, string> _connectionDetails;
private readonly IDictionary<string, string> _primaryConfig;
private readonly IDictionary<string, string>? _backupConfig;
private readonly int _failoverRetryCount;
// The three fields below describe the configuration currently in force. None are readonly:
// an in-place reconfiguration (ReconfigureConnectionCommand, deferred register row 97)
// replaces them wholesale and then swaps the adapter through the same machinery the
// primary→backup failover uses.
private IDictionary<string, string> _primaryConfig;
private IDictionary<string, string>? _backupConfig;
private int _failoverRetryCount;
private ActiveEndpoint _activeEndpoint = ActiveEndpoint.Primary;
private int _consecutiveFailures;
private int _consecutiveUnstableDisconnects;
@@ -339,6 +348,13 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers
case UnsubscribeAlarmsRequest:
Stash.Stash();
break;
case ReconfigureConnectionCommand rc:
// NEVER stashed. A reconfigure that arrived while the initial connect was still
// in flight would otherwise sit in the stash until the OLD configuration
// succeeded — and if it never succeeds (the very case an endpoint correction is
// sent to fix) it would sit there forever.
HandleReconfigure(rc);
break;
case SubscribeCompleted sc:
// A subscribe started while Connected can complete after a transition;
// apply it so its state survives into the next ReSubscribeAll.
@@ -454,6 +470,9 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers
case AdapterDisconnected:
HandleDisconnect();
break;
case ReconfigureConnectionCommand rc:
HandleReconfigure(rc);
break;
case RetryTagResolution:
HandleRetryTagResolution();
break;
@@ -477,67 +496,65 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers
// ── Reconnecting State ──
private void BecomeReconnecting()
/// <param name="afterReconfigure">
/// True when this transition was caused by an in-place reconfiguration
/// (<see cref="HandleReconfigure"/>) rather than by losing the link. It suppresses the
/// unstable-disconnect accounting and the failover check — the adapter has just been
/// swapped ON PURPOSE against a brand-new configuration, so counting it as an unstable
/// cycle would let a reconfigure trip a failover onto the new backup before the new primary
/// has been tried even once — and it reconnects immediately instead of waiting out
/// <see cref="DataConnectionOptions.ReconnectInterval"/>, since there is nothing to wait for.
/// </param>
private void BecomeReconnecting(bool afterReconfigure = false)
{
_log.Warning("[{0}] Entering Reconnecting state", _connectionName);
_log.Warning("[{0}] Entering Reconnecting state{1}", _connectionName,
afterReconfigure ? " (reconfigured in place)" : string.Empty);
_healthCollector.UpdateConnectionHealth(_connectionName, ConnectionHealth.Disconnected);
_healthCollector.UpdateConnectionEndpoint(_connectionName, "Disconnected");
// Register row 54 — drops this connection out of the connected gauge while the
// total holds, which is exactly the connected < total alert condition.
ScadaBridgeTelemetry.RecordDataConnectionState(_connectionName, ConnectionHealth.Disconnected);
// Track unstable connections toward failover.
// If we were connected for less than the stability threshold, this counts
// as an unstable cycle (e.g., connect succeeded but heartbeat went stale).
var connectionDuration = DateTimeOffset.UtcNow - _lastConnectedAt;
if (_lastConnectedAt != default && connectionDuration < _options.StableConnectionThreshold)
if (!afterReconfigure)
{
_consecutiveUnstableDisconnects++;
_log.Warning("[{0}] Unstable connection (lasted {1:F0}s) — consecutive unstable disconnects: {2}/{3}",
_connectionName, connectionDuration.TotalSeconds, _consecutiveUnstableDisconnects,
_backupConfig != null ? _failoverRetryCount : 0);
}
else
{
_consecutiveUnstableDisconnects = 0;
}
// Failover if we keep connecting and going stale repeatedly
if (_backupConfig != null && _consecutiveUnstableDisconnects >= _failoverRetryCount)
{
var previousEndpoint = _activeEndpoint;
_activeEndpoint = _activeEndpoint == ActiveEndpoint.Primary
? ActiveEndpoint.Backup
: ActiveEndpoint.Primary;
_consecutiveUnstableDisconnects = 0;
_consecutiveFailures = 0;
var newConfig = _activeEndpoint == ActiveEndpoint.Primary
? _primaryConfig
: _backupConfig;
// Dispose old adapter
_adapter.Disconnected -= OnAdapterDisconnected;
_ = _adapter.DisposeAsync().AsTask();
// Create new adapter for the target endpoint
_adapter = _factory.Create(_protocolType, newConfig);
_connectionDetails = newConfig;
_adapter.Disconnected += OnAdapterDisconnected;
// New adapter — bump the generation so callbacks
// from the disposed adapter are recognised as stale and dropped.
_adapterGeneration++;
_log.Warning("[{0}] Failing over from {1} to {2} (unstable connection pattern)",
_connectionName, previousEndpoint, _activeEndpoint);
if (_siteEventLogger != null)
// Track unstable connections toward failover.
// If we were connected for less than the stability threshold, this counts
// as an unstable cycle (e.g., connect succeeded but heartbeat went stale).
var connectionDuration = DateTimeOffset.UtcNow - _lastConnectedAt;
if (_lastConnectedAt != default && connectionDuration < _options.StableConnectionThreshold)
{
_ = _siteEventLogger.LogEventAsync(
"connection", "Warning", null, _connectionName,
$"Failover from {previousEndpoint} to {_activeEndpoint} (unstable connection)",
$"Connection lasted {connectionDuration.TotalSeconds:F0}s, threshold {_options.StableConnectionThreshold.TotalSeconds:F0}s");
_consecutiveUnstableDisconnects++;
_log.Warning("[{0}] Unstable connection (lasted {1:F0}s) — consecutive unstable disconnects: {2}/{3}",
_connectionName, connectionDuration.TotalSeconds, _consecutiveUnstableDisconnects,
_backupConfig != null ? _failoverRetryCount : 0);
}
else
{
_consecutiveUnstableDisconnects = 0;
}
// Failover if we keep connecting and going stale repeatedly
if (_backupConfig != null && _consecutiveUnstableDisconnects >= _failoverRetryCount)
{
var previousEndpoint = _activeEndpoint;
_activeEndpoint = _activeEndpoint == ActiveEndpoint.Primary
? ActiveEndpoint.Backup
: ActiveEndpoint.Primary;
_consecutiveUnstableDisconnects = 0;
_consecutiveFailures = 0;
SwapAdapter(_activeEndpoint == ActiveEndpoint.Primary ? _primaryConfig : _backupConfig);
_log.Warning("[{0}] Failing over from {1} to {2} (unstable connection pattern)",
_connectionName, previousEndpoint, _activeEndpoint);
if (_siteEventLogger != null)
{
_ = _siteEventLogger.LogEventAsync(
"connection", "Warning", null, _connectionName,
$"Failover from {previousEndpoint} to {_activeEndpoint} (unstable connection)",
$"Connection lasted {connectionDuration.TotalSeconds:F0}s, threshold {_options.StableConnectionThreshold.TotalSeconds:F0}s");
}
}
}
@@ -545,8 +562,11 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers
if (_siteEventLogger != null)
{
_ = _siteEventLogger.LogEventAsync(
"connection", "Warning", null, _connectionName,
$"Connection lost — entering reconnect cycle", null);
"connection", afterReconfigure ? "Info" : "Warning", null, _connectionName,
afterReconfigure
? "Connection reconfigured — reconnecting on the new configuration"
: "Connection lost — entering reconnect cycle",
null);
}
Become(Reconnecting);
@@ -558,8 +578,118 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers
// (mark mirrored alarms uncertain; the reconnect snapshot reconciles them).
PushAlarmSourceUnavailable();
// Schedule reconnect attempt
Timers.StartSingleTimer("reconnect", new AttemptConnect(), _options.ReconnectInterval);
// Schedule reconnect attempt. StartSingleTimer REPLACES any reconnect already armed,
// so a reconfigure landing mid-backoff cannot leave two AttemptConnect messages racing.
Timers.StartSingleTimer("reconnect", new AttemptConnect(),
afterReconfigure ? TimeSpan.Zero : _options.ReconnectInterval);
}
/// <summary>
/// Replaces the live adapter with one built for <paramref name="newConfig"/> — the single
/// swap shared by both failover sites and by in-place reconfiguration
/// (<see cref="HandleReconfigure"/>).
///
/// <para>
/// The old adapter is unhooked first and disposed fire-and-forget (never awaited on the
/// actor thread), and <see cref="_adapterGeneration"/> is bumped LAST so every callback and
/// in-flight background task started against the outgoing adapter is recognised as stale and
/// dropped rather than delivered as if it came from the new endpoint. Subscriptions are
/// untouched: they live in <see cref="_subscriptionsByInstance"/> and are re-issued by
/// <see cref="ReSubscribeAll"/> once the replacement adapter connects.
/// </para>
/// </summary>
private void SwapAdapter(IDictionary<string, string> newConfig)
{
_adapter.Disconnected -= OnAdapterDisconnected;
_ = _adapter.DisposeAsync().AsTask();
_adapter = _factory.Create(_protocolType, newConfig);
_connectionDetails = newConfig;
_adapter.Disconnected += OnAdapterDisconnected;
_adapterGeneration++;
}
/// <summary>
/// Applies a new configuration to this LIVE connection <b>in place</b> (deferred register
/// row 97) — the non-destructive alternative to stopping and re-creating the connection
/// actor, which would strand every subscriber (the subscription registry lives here, and
/// <c>InstanceActor</c> cancels its subscribe-retry timer on the first success, so nothing
/// would ever re-subscribe).
///
/// <para>
/// It reuses the primary→backup failover machinery verbatim: <see cref="SwapAdapter"/>
/// disposes the current adapter, builds one from the new configuration and bumps the
/// generation, then the reconnect cycle re-issues every subscription from
/// <see cref="_subscriptionsByInstance"/> and every alarm feed from
/// <see cref="_alarmSourceSubscribers"/>. In-flight writes against the outgoing adapter may
/// fail — the same accepted consequence a failover has, and the failure is returned
/// synchronously to the calling script exactly as it would be there.
/// </para>
///
/// <para>
/// <b>Endpoint selection is re-armed from the top.</b> Whatever endpoint the PREVIOUS
/// configuration had failed over to, the new configuration's primary is the operator's
/// intent, so the active endpoint and both failure counters reset.
/// </para>
///
/// <para>
/// <b>Idempotent.</b> A reconfigure carrying the configuration already in force is a no-op.
/// The sender is hash-gated and does not send one, but a swap is destructive enough that a
/// duplicate must never cost a reconnect.
/// </para>
/// </summary>
private void HandleReconfigure(ReconfigureConnectionCommand command)
{
if (string.Equals(_protocolType, command.ProtocolType, StringComparison.Ordinal)
&& _failoverRetryCount == command.FailoverRetryCount
&& DetailsEqual(_primaryConfig, command.PrimaryConnectionDetails)
&& DetailsEqual(_backupConfig, command.BackupConnectionDetails))
{
_log.Debug("[{0}] Reconfigure carries the configuration already in force — ignoring",
_connectionName);
return;
}
_log.Warning(
"[{0}] Reconfiguring in place (protocol {1} → {2}) — swapping the adapter and "
+ "re-subscribing {3} tag(s) for {4} instance(s) and {5} alarm source(s)",
_connectionName, _protocolType, command.ProtocolType,
_instancesByTag.Count, _subscriptionsByInstance.Count, _alarmSourceSubscribers.Count);
_protocolType = command.ProtocolType;
_primaryConfig = command.PrimaryConnectionDetails;
_backupConfig = command.BackupConnectionDetails;
_failoverRetryCount = command.FailoverRetryCount;
_activeEndpoint = ActiveEndpoint.Primary;
_consecutiveFailures = 0;
_consecutiveUnstableDisconnects = 0;
SwapAdapter(_primaryConfig);
BecomeReconnecting(afterReconfigure: true);
}
/// <summary>
/// Order-insensitive value comparison of two flattened connection-detail maps, used only by
/// <see cref="HandleReconfigure"/>'s no-op guard.
/// </summary>
private static bool DetailsEqual(
IDictionary<string, string>? left, IDictionary<string, string>? right)
{
if (ReferenceEquals(left, right)) return true;
if (left == null || right == null) return false;
if (left.Count != right.Count) return false;
foreach (var (key, value) in left)
{
if (!right.TryGetValue(key, out var other)
|| !string.Equals(value, other, StringComparison.Ordinal))
return false;
}
return true;
}
private void Reconnecting(object message)
@@ -586,6 +716,11 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers
// Allow alarm unsubscribe during reconnect (cleanup on instance stop).
HandleUnsubscribeAlarms(areq);
break;
case ReconfigureConnectionCommand rc:
// NEVER stashed — this is the state a connection to a wrong or dead endpoint
// spends its life in, i.e. exactly when a corrected configuration arrives.
HandleReconfigure(rc);
break;
case TagValueReceived:
// Ignore — stale callback from previous connection
break;
@@ -720,24 +855,7 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers
: ActiveEndpoint.Primary;
_consecutiveFailures = 0;
var newConfig = _activeEndpoint == ActiveEndpoint.Primary
? _primaryConfig
: _backupConfig;
// Dispose old adapter (fire-and-forget — don't await in actor context)
_adapter.Disconnected -= OnAdapterDisconnected;
_ = _adapter.DisposeAsync().AsTask();
// Create new adapter for the target endpoint
_adapter = _factory.Create(_protocolType, newConfig);
_connectionDetails = newConfig;
// Wire disconnect handler on new adapter
_adapter.Disconnected += OnAdapterDisconnected;
// New adapter — bump the generation so callbacks
// from the disposed adapter are recognised as stale and dropped.
_adapterGeneration++;
SwapAdapter(_activeEndpoint == ActiveEndpoint.Primary ? _primaryConfig : _backupConfig);
_log.Warning("[{0}] Failing over from {1} to {2}",
_connectionName, previousEndpoint, _activeEndpoint);
src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Actors/DataConnectionManagerActor.cs
+35
View File
@@ -58,6 +58,7 @@ public class DataConnectionManagerActor : ReceiveActor
_opcUaGlobalOptions = opcUaGlobalOptions ?? new OpcUaGlobalOptions();
Receive<CreateConnectionCommand>(HandleCreateConnection);
Receive<ReconfigureConnectionCommand>(HandleReconfigureConnection);
Receive<SubscribeTagsRequest>(HandleRoute);
Receive<UnsubscribeTagsRequest>(HandleRoute);
Receive<SubscribeAlarmsRequest>(HandleRouteAlarms);
@@ -102,6 +103,40 @@ public class DataConnectionManagerActor : ReceiveActor
command.ConnectionName, command.ProtocolType);
}
/// <summary>
/// Applies a new configuration to a LIVE connection actor (deferred register row 97) by
/// forwarding the command to it — the actor swaps its adapter in place and re-issues every
/// subscription it holds, rather than being stopped and rebuilt (which would silently strand
/// every subscriber; see <see cref="ReconfigureConnectionCommand"/>).
///
/// <para>
/// A name this manager does not hold falls back to <see cref="HandleCreateConnection"/>, so
/// the command is safe regardless of manager state: a sender whose view of which connections
/// exist is stale (a fresh singleton incarnation, a node that has just taken over) still
/// converges on the intended configuration in one message.
/// </para>
/// </summary>
private void HandleReconfigureConnection(ReconfigureConnectionCommand command)
{
if (_connectionActors.TryGetValue(command.ConnectionName, out var actor))
{
_log.Info("Reconfiguring DataConnectionActor for {0} in place (protocol={1})",
command.ConnectionName, command.ProtocolType);
actor.Forward(command);
return;
}
_log.Info(
"Reconfigure for {0} arrived with no live connection actor — creating it instead",
command.ConnectionName);
HandleCreateConnection(new CreateConnectionCommand(
command.ConnectionName,
command.ProtocolType,
command.PrimaryConnectionDetails,
command.BackupConnectionDetails,
command.FailoverRetryCount));
}
private void HandleRoute(SubscribeTagsRequest request)
{
if (_connectionActors.TryGetValue(request.ConnectionName, out var actor))