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

fix(dcl): capture adapter into locals before background tasks — no off-thread reads of the mutable _adapter field (S7)

Browse Source
This commit is contained in:
Joseph Doherty
2026-07-09 00:44:09 -04:00
parent ef69fab6f6
commit f294b46263
2 changed files with 97 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Actors/DataConnectionActor.cs
+26 -6
View File
@@ -678,6 +678,13 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers
// Capture the current adapter generation so callbacks
// from this adapter can be distinguished from a later (post-failover) adapter.
var generation = _adapterGeneration;
// Capture the adapter reference on the actor thread too (S7). The background task
// below iterates multiple tags with an await between each SubscribeAsync — a failover
// that swaps the mutable _adapter field mid-loop would otherwise make later iterations
// (and the SeedTagsAsync read) run against the NEW adapter while carrying the OLD
// generation. The in-flight subscribe must finish against the adapter that started it;
// the generation guard drops any value the swapped-out adapter later produces.
var adapter = _adapter;
// Partition tags on the actor thread into "this
// request will issue _adapter.SubscribeAsync" vs. "already subscribed (by us
@@ -718,7 +725,7 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers
{
try
{
var subId = await _adapter.SubscribeAsync(tagPath, (path, value) =>
var subId = await adapter.SubscribeAsync(tagPath, (path, value) =>
{
self.Tell(new TagValueReceived(path, value, generation));
});
@@ -752,7 +759,7 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers
// SeedTagsAsync retries the still-empty reads so a
// seed that races the just-created advise (returns VT_EMPTY) is not silently
// dropped, and logs any tag that never yields a value.
var seedValues = await SeedTagsAsync(_adapter, tagsToSeed);
var seedValues = await SeedTagsAsync(adapter, tagsToSeed);
return new SubscribeCompleted(request, sender, results, seedValues);
}).PipeTo(self);
@@ -1119,8 +1126,12 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers
// failed WriteTagResponse so the failure is returned synchronously.
var cts = new CancellationTokenSource(_options.WriteTimeout);
// Capture the adapter on the actor thread (S7) so the write completes against the
// adapter that started it even if a failover swaps the _adapter field mid-flight.
var adapter = _adapter;
// Write through DCL to device, failure returned synchronously
_adapter.WriteAsync(request.TagPath, request.Value, cts.Token).ContinueWith(t =>
adapter.WriteAsync(request.TagPath, request.Value, cts.Token).ContinueWith(t =>
{
cts.Dispose();
if (t.IsCompletedSuccessfully)
@@ -1158,11 +1169,16 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers
var sender = Sender;
var cts = new CancellationTokenSource(_options.WriteTimeout);
// Capture the adapter on the actor thread (S7). The trigger write below runs AFTER
// the batch await, so re-reading the _adapter field there could hit a post-failover
// adapter; both legs must use the adapter that started this batch.
var adapter = _adapter;
async Task<WriteTagBatchResponse> RunAsync()
{
try
{
var results = await _adapter.WriteBatchAsync(
var results = await adapter.WriteBatchAsync(
new Dictionary<string, object?>(request.Values), cts.Token);
var failed = results.Values.Where(r => !r.Success).Select(r => r.ErrorMessage).ToList();
if (failed.Count > 0)
@@ -1172,7 +1188,7 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers
if (!string.IsNullOrEmpty(request.TriggerTagPath))
{
var tr = await _adapter.WriteAsync(request.TriggerTagPath, request.TriggerValue, cts.Token);
var tr = await adapter.WriteAsync(request.TriggerTagPath, request.TriggerValue, cts.Token);
if (!tr.Success)
return new WriteTagBatchResponse(
request.CorrelationId, false,
@@ -1591,9 +1607,13 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers
_healthCollector.UpdateTagQuality(_connectionName, _tagsGoodQuality, _tagsBadQuality, _tagsUncertainQuality);
var generation = _adapterGeneration;
// Capture the adapter up front (S7), symmetric with reseedAdapter below. This loop
// runs on the actor thread so a mid-loop swap can't happen, but reading the local
// keeps every adapter access in this method tied to one generation.
var subscribeAdapter = _adapter;
foreach (var tagPath in allTags)
{
_adapter.SubscribeAsync(tagPath, (path, value) =>
subscribeAdapter.SubscribeAsync(tagPath, (path, value) =>
{
self.Tell(new TagValueReceived(path, value, generation));
}).ContinueWith(t =>
tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/DataConnectionActorTests.cs
+71
View File
@@ -463,6 +463,77 @@ public class DataConnectionActorTests : TestKit
"sensor/temp", Arg.Any<SubscriptionCallback>(), Arg.Any<CancellationToken>());
}
// ── PLAN-03 Task 16 (S7): background tasks must use the adapter that started them ──
[Fact]
public async Task Task16_InFlightSubscribe_ContinuesOnOriginalAdapter_AfterFailover()
{
// Regression for S7. HandleSubscribe launches a background Task.Run that iterates
// the tags and awaits _adapter.SubscribeAsync per tag. If a failover swaps the
// mutable _adapter field mid-loop, later iterations must still hit the adapter that
// STARTED the subscribe — not the freshly-swapped-in backup. Pre-fix the loop re-read
// the _adapter field after the first await, so a tag after the first was subscribed
// against the wrong (backup) adapter generation.
var primaryConfig = new Dictionary<string, string> { ["Endpoint"] = "opc.tcp://primary:4840" };
var backupConfig = new Dictionary<string, string> { ["Endpoint"] = "opc.tcp://backup:4840" };
var primaryAdapter = Substitute.For<IDataConnection>();
var backupAdapter = Substitute.For<IDataConnection>();
// Primary connects once, then every reconnect fails so the disconnect drives failover.
var primaryConnectCount = 0;
primaryAdapter.ConnectAsync(Arg.Any<IDictionary<string, string>>(), Arg.Any<CancellationToken>())
.Returns(_ => Interlocked.Increment(ref primaryConnectCount) == 1
? Task.CompletedTask
: Task.FromException(new Exception("Primary down")));
// The first tag's subscribe blocks on the gate (in-flight across the failover);
// the second tag resolves immediately once the loop resumes.
var subscribeGate = new TaskCompletionSource<string>(TaskCreationOptions.RunContinuationsAsynchronously);
primaryAdapter.SubscribeAsync(Arg.Any<string>(), Arg.Any<SubscriptionCallback>(), Arg.Any<CancellationToken>())
.Returns(_ => subscribeGate.Task, _ => Task.FromResult("sub-primary-2"));
primaryAdapter.ReadAsync(Arg.Any<string>(), Arg.Any<CancellationToken>())
.Returns(new ReadResult(false, null, null));
_mockFactory.Create("OpcUa", Arg.Is<IDictionary<string, string>>(d => d["Endpoint"] == "opc.tcp://backup:4840"))
.Returns(backupAdapter);
backupAdapter.ConnectAsync(Arg.Any<IDictionary<string, string>>(), Arg.Any<CancellationToken>())
.Returns(Task.CompletedTask);
backupAdapter.SubscribeAsync(Arg.Any<string>(), Arg.Any<SubscriptionCallback>(), Arg.Any<CancellationToken>())
.Returns("sub-backup");
backupAdapter.ReadAsync(Arg.Any<string>(), Arg.Any<CancellationToken>())
.Returns(new ReadResult(false, null, null));
var actor = CreateFailoverActor(primaryAdapter, "task16-inflight", primaryConfig, backupConfig, failoverRetryCount: 1);
// Reach Connected on primary.
AwaitCondition(() => primaryConnectCount >= 1, TimeSpan.FromSeconds(2));
await Task.Delay(200);
// Kick off a two-tag subscribe; the first tag blocks in-flight on the gate.
actor.Tell(new SubscribeTagsRequest("c1", "inst1", "task16-inflight", ["t1", "t2"], DateTimeOffset.UtcNow));
// The background loop must have called the primary's SubscribeAsync (t1) and parked.
AwaitCondition(() =>
primaryAdapter.ReceivedCalls().Count(c => c.GetMethodInfo().Name == "SubscribeAsync") >= 1,
TimeSpan.FromSeconds(3));
// Fail over while the subscribe is in flight: swaps _adapter to backup, bumps generation.
RaiseDisconnected(primaryAdapter);
AwaitCondition(() =>
backupAdapter.ReceivedCalls().Any(c => c.GetMethodInfo().Name == "ConnectAsync"),
TimeSpan.FromSeconds(5));
// Release the in-flight subscribe; the loop resumes and subscribes t2.
subscribeGate.SetResult("sub-primary-1");
// t2 must go to the ORIGINAL (primary) adapter, never the swapped-in backup.
AwaitCondition(() =>
primaryAdapter.ReceivedCalls().Count(c => c.GetMethodInfo().Name == "SubscribeAsync") >= 2,
TimeSpan.FromSeconds(5));
Assert.DoesNotContain(backupAdapter.ReceivedCalls(),
c => c.GetMethodInfo().Name == "SubscribeAsync");
}
// ── DataConnectionLayer-001: subscribe must not mutate actor state off-thread ──
private static async Task<string> DelayedSubscribeAsync()