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

fix(dcl): carry adapter generation on SubscribeCompleted/AlarmSubscribeCompleted; drop stale completions and release captured-adapter handles (arch-review 2026-08-19 P2-1 High)

Browse Source 
An in-flight subscribe launched against adapter generation N can take up to
SeedOverallTimeout (120s) to answer. If the adapter is swapped in that window
(failover, or the row-97 in-place reconfigure), the completion was applied
unconditionally: it stored a dead session's handle under the tag key, so the
LIVE completion for the same tag read as a duplicate and had its working
monitored item released (tag resolved forever, no data); it re-stamped the old
endpoint's seed values with the CURRENT generation, bypassing the stale-data
guard; and a connection-level failure row from the dead adapter knocked the
healthy new connection back into Reconnecting. The alarm twin released the new
live alarm feed the same way.

Both completions now carry the generation AND the adapter instance they ran
against, and both handlers drop a completion whose generation no longer
matches - no handle store, no seed application, no per-tag or connection-level
failure processing, no duplicate-release logic. Handles the dropped completion
did create are released best-effort against the CAPTURED adapter (handles are
adapter-scoped; MxGateway's are plain ints that collide across sessions), and
the existing release sites in both handlers were switched to the message's
adapter for the same reason. The in-flight markers are deliberately left alone
(ReSubscribeAll/ReSubscribeAllAlarms clear them wholesale and may already have
re-registered the tag/source), and the requester gets a failure reply so its
retry timer re-sends.

Tests: DataConnectionStaleCompletionTests - stale handle not stored / live item
not released, release lands on the old adapter only, seed values not delivered,
stale connection-level failure does not re-enter Reconnecting, stale alarm
completion does not release the live feed. All five fail against the unfixed
handlers.
This commit is contained in:
Joseph Doherty
2026-08-19 13:54:06 -04:00
parent e9e44a9ef0
commit c4d3c8b875
2 changed files with 500 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Actors/DataConnectionActor.cs
+140 -21
View File
@@ -983,7 +983,7 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers
// dropped, and logs any tag that never yields a value.
var seedValues = await SeedTagsAsync(adapter, tagsToSeed);
return new SubscribeCompleted(request, sender, results, seedValues);
return new SubscribeCompleted(request, sender, results, seedValues, generation, adapter);
}).PipeTo(self);
}
@@ -1086,6 +1086,30 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers
return Task.WhenAll(subscriptionIds.Select(id => adapter.UnsubscribeAsync(id)));
}
/// <summary>
/// Best-effort release of handles created by a subscribe whose adapter has since been
/// swapped out. <paramref name="release"/> must target the adapter that CREATED the handles
/// — they are adapter-scoped (MxGateway's are plain ints that collide across sessions), so
/// releasing by handle against the replacement adapter can tear down a live subscription.
/// The outgoing adapter is disposed fire-and-forget by <see cref="SwapAdapter"/> and may
/// already be gone, so a failure here is expected and logged at Debug rather than surfaced.
/// </summary>
private void ReleaseStaleAgainstOwnAdapter(string description, Func<Task> release)
{
try
{
release().ContinueWith(
t => _log.Debug("[{0}] Best-effort release of {1} failed: {2}",
_connectionName, description, t.Exception?.GetBaseException().Message),
TaskContinuationOptions.OnlyOnFaulted);
}
catch (Exception ex)
{
_log.Debug("[{0}] Best-effort release of {1} threw: {2}",
_connectionName, description, ex.Message);
}
}
/// <summary>
/// Reads the current value of each tag so the Instance Actor
/// has an initial value, retrying the still-empty subset a bounded number of times. A
@@ -1256,6 +1280,51 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers
private bool HandleSubscribeCompleted(SubscribeCompleted msg)
{
var instanceName = msg.Request.InstanceUniqueName;
// Results produced against an adapter that has since been swapped out (failover or
// in-place reconfigure) are dropped WHOLE, mirroring the BatchSubscribeCompleted and
// TagValueReceived generation guards. An initial subscribe can take up to
// SeedOverallTimeout to complete, so this window is wide. Applying a stale completion
// stores a dead session's handle under the tag's key, which then makes the LIVE
// completion for the same tag look like a duplicate and get RELEASED (the tag reads
// resolved forever and delivers nothing); re-stamps the old endpoint's seed values with
// the CURRENT generation, so they bypass the stale-data guard and present as fresh; and
// lets a connection-level failure row from the dead adapter knock the healthy new
// connection back into Reconnecting.
if (msg.AdapterGeneration != _adapterGeneration)
{
var staleIds = msg.Results
.Where(r => r is { Success: true, AlreadySubscribed: false, SubscriptionId: not null })
.Select(r => r.SubscriptionId!)
.ToList();
_log.Debug(
"[{0}] Dropping stale subscribe completion for instance {1} from adapter " +
"generation {2} (current {3}); releasing {4} handle(s) against the adapter that created them",
_connectionName, instanceName, msg.AdapterGeneration, _adapterGeneration, staleIds.Count);
// Release against msg.Adapter, NEVER _adapter: handles are adapter-scoped and
// MxGateway's are plain ints that collide across sessions, so releasing by handle
// against the replacement adapter can tear down a live subscription.
ReleaseStaleAgainstOwnAdapter(
$"{staleIds.Count} tag subscription handle(s) from adapter generation {msg.AdapterGeneration}",
() => UnsubscribeIdsAsync(msg.Adapter, staleIds));
// _subscribesInFlight is deliberately NOT cleared here: ReSubscribeAll empties it
// wholesale once the replacement adapter connects, and a marker the new generation
// has since placed for the same tag must not be removed by this dead message.
// Reply failure so the requester re-sends. Nothing has registered this request's
// tags in _subscriptionsByInstance (registration happens below, on the applied
// path), so ReSubscribeAll cannot pick them up — the InstanceActor's per-connection
// retry timer, armed until a Success arrives, is what re-establishes them.
msg.ReplyTo.Tell(new SubscribeTagsResponse(
msg.Request.CorrelationId, instanceName, false,
"adapter replaced while the subscribe was in flight — will retry",
DateTimeOffset.UtcNow));
return false;
}
if (!_subscriptionsByInstance.TryGetValue(instanceName, out var instanceTags))
{
// The instance was unsubscribed while the
@@ -1292,7 +1361,10 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers
}
}
_ = UnsubscribeIdsAsync(_adapter, orphanedIds);
// Against the adapter that created them (identical to _adapter here — the
// generation guard above already proved the completion is current — but stating
// the ownership rule at every release site is what keeps it true).
_ = UnsubscribeIdsAsync(msg.Adapter, orphanedIds);
// The original sender is already gone (unsubscribed). Telling a dead
// ref produces a dead letter, which is the harmless and observable
@@ -1400,8 +1472,9 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers
}
}
// Fire-and-forget release of every handle this request created redundantly.
_ = UnsubscribeIdsAsync(_adapter, redundantIds);
// Fire-and-forget release of every handle this request created redundantly, against the
// adapter that created them (see the orphaned-release site above).
_ = UnsubscribeIdsAsync(msg.Adapter, redundantIds);
// Publish the resolution counts this request just changed. Without this push a
// connection whose tags ALL resolve on the first attempt reported 0/0 forever:
@@ -2502,14 +2575,46 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers
alarmable.SubscribeAlarmsAsync(sourceRef, filter,
t => self.Tell(new AlarmTransitionReceived(t, generation)))
.ContinueWith(task => task.IsCompletedSuccessfully
? new AlarmSubscribeCompleted(sourceRef, true, task.Result, null, subscriber, corr, inst) as object
? new AlarmSubscribeCompleted(
sourceRef, true, task.Result, null, subscriber, corr, inst, generation, alarmable) as object
: new AlarmSubscribeCompleted(sourceRef, false, null,
task.Exception?.GetBaseException().Message ?? "Unknown error", subscriber, corr, inst))
task.Exception?.GetBaseException().Message ?? "Unknown error",
subscriber, corr, inst, generation, alarmable))
.PipeTo(self);
}
private void HandleAlarmSubscribeCompleted(AlarmSubscribeCompleted msg)
{
// The alarm twin of HandleSubscribeCompleted's generation guard: a completion from an
// adapter that has since been swapped out is dropped whole. Applying it would store a
// dead feed's id — or, when the new adapter's feed is already stored, take the
// duplicate branch and release the LIVE feed, leaving the source mirrored but silent.
if (msg.AdapterGeneration != _adapterGeneration)
{
_log.Debug(
"[{0}] Dropping stale alarm subscribe completion for source {1} from adapter " +
"generation {2} (current {3})",
_connectionName, msg.SourceReference, msg.AdapterGeneration, _adapterGeneration);
// Release the orphaned feed against the adapter that created it, never the current
// one (see ReleaseStaleAgainstOwnAdapter).
if (msg is { Success: true, SubscriptionId: not null })
ReleaseStaleAgainstOwnAdapter(
$"the alarm feed for {msg.SourceReference} from adapter generation {msg.AdapterGeneration}",
() => msg.Adapter.UnsubscribeAlarmsAsync(msg.SubscriptionId));
// _alarmSubscribesInFlight is left untouched — ReSubscribeAllAlarms clears it
// wholesale and may already have re-registered this source against the new adapter.
// The subscriber state (_alarmSourceSubscribers) survives the swap, so the feed is
// re-established by ReSubscribeAllAlarms; the failure reply just lets the
// NativeAlarmActor retry rather than wait out its response timeout.
msg.ReplyTo?.Tell(new SubscribeAlarmsResponse(
msg.CorrelationId ?? string.Empty, msg.InstanceUniqueName ?? string.Empty,
false, "adapter replaced while the alarm subscribe was in flight — will retry",
DateTimeOffset.UtcNow));
return;
}
_alarmSubscribesInFlight.Remove(msg.SourceReference);
// The last (or only) subscriber may have been
@@ -2521,15 +2626,16 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers
// transitions to nobody for the lifetime of the adapter.
if (!_alarmSourceSubscribers.ContainsKey(msg.SourceReference))
{
if (msg.Success && msg.SubscriptionId != null &&
_adapter is IAlarmSubscribableConnection alarmable)
if (msg.Success && msg.SubscriptionId != null)
{
_log.Warning(
"[{0}] AlarmSubscribeCompleted arrived for source {1} but the last " +
"subscriber unsubscribed while the subscribe was in flight; releasing " +
"the orphaned adapter alarm feed.",
_connectionName, msg.SourceReference);
_ = alarmable.UnsubscribeAlarmsAsync(msg.SubscriptionId);
// Against the adapter that created the feed (identical to _adapter here — the
// generation guard above already proved the completion is current).
_ = msg.Adapter.UnsubscribeAlarmsAsync(msg.SubscriptionId);
}
// No live requester remains to receive a response.
@@ -2547,14 +2653,11 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers
// the stored id and leaking the already-tracked subscription.
if (_alarmSubscriptionIds.ContainsKey(msg.SourceReference))
{
if (_adapter is IAlarmSubscribableConnection alarmable)
{
_log.Warning(
"[{0}] Duplicate alarm feed for source {1}; releasing the redundant " +
"subscription instead of overwriting the stored one.",
_connectionName, msg.SourceReference);
_ = alarmable.UnsubscribeAlarmsAsync(msg.SubscriptionId);
}
_log.Warning(
"[{0}] Duplicate alarm feed for source {1}; releasing the redundant " +
"subscription instead of overwriting the stored one.",
_connectionName, msg.SourceReference);
_ = msg.Adapter.UnsubscribeAlarmsAsync(msg.SubscriptionId);
}
else
{
@@ -2755,9 +2858,11 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers
alarmable.SubscribeAlarmsAsync(sr, filter,
t => self.Tell(new AlarmTransitionReceived(t, generation)))
.ContinueWith(task => task.IsCompletedSuccessfully
? new AlarmSubscribeCompleted(sr, true, task.Result, null, null, null, null) as object
? new AlarmSubscribeCompleted(
sr, true, task.Result, null, null, null, null, generation, alarmable) as object
: new AlarmSubscribeCompleted(sr, false, null,
task.Exception?.GetBaseException().Message ?? "Unknown error", null, null, null))
task.Exception?.GetBaseException().Message ?? "Unknown error",
null, null, null, generation, alarmable))
.PipeTo(self);
}
}
@@ -2803,16 +2908,30 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers
internal record SubscribeTagResult(
string TagPath, bool AlreadySubscribed, bool Success, string? SubscriptionId, string? Error,
bool ConnectionLevelFailure = false);
/// <summary>
/// Per-tag results of the initial subscribe issued off the actor thread by
/// <see cref="HandleSubscribe"/>. Carries the adapter generation it ran against — and the
/// adapter INSTANCE that produced the handles, because a stale completion's handles must be
/// released against their own adapter (see <see cref="HandleSubscribeCompleted"/>).
/// </summary>
internal record SubscribeCompleted(
SubscribeTagsRequest Request, IActorRef ReplyTo, IReadOnlyList<SubscribeTagResult> Results,
IReadOnlyList<SeededValue> SeedValues);
IReadOnlyList<SeededValue> SeedValues, int AdapterGeneration, IDataConnection Adapter);
/// <summary>An initial-read value captured during subscribe, delivered after the
/// instance's tags are registered for fan-out.</summary>
internal record SeededValue(string TagPath, TagValue Value);
internal record AlarmTransitionReceived(NativeAlarmTransition Transition, int AdapterGeneration);
/// <summary>
/// Result of an alarm-feed subscribe issued off the actor thread. Carries the adapter
/// generation and the adapter instance for the same reason
/// <see cref="SubscribeCompleted"/> does.
/// </summary>
internal record AlarmSubscribeCompleted(
string SourceReference, bool Success, string? SubscriptionId, string? Error,
IActorRef? ReplyTo, string? CorrelationId, string? InstanceUniqueName);
IActorRef? ReplyTo, string? CorrelationId, string? InstanceUniqueName,
int AdapterGeneration, IAlarmSubscribableConnection Adapter);
public record GetHealthReport;
}