From f294b46263633d8e9bd41ad890a140e8a26ebfac Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 9 Jul 2026 00:44:09 -0400 Subject: [PATCH] =?UTF-8?q?fix(dcl):=20capture=20adapter=20into=20locals?= =?UTF-8?q?=20before=20background=20tasks=20=E2=80=94=20no=20off-thread=20?= =?UTF-8?q?reads=20of=20the=20mutable=20=5Fadapter=20field=20(S7)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../Actors/DataConnectionActor.cs | 32 +++++++-- .../DataConnectionActorTests.cs | 71 +++++++++++++++++++ 2 files changed, 97 insertions(+), 6 deletions(-) diff --git a/src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Actors/DataConnectionActor.cs b/src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Actors/DataConnectionActor.cs index 8e906139..32ab75f7 100644 --- a/src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Actors/DataConnectionActor.cs +++ b/src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Actors/DataConnectionActor.cs @@ -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 RunAsync() { try { - var results = await _adapter.WriteBatchAsync( + var results = await adapter.WriteBatchAsync( new Dictionary(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 => diff --git a/tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/DataConnectionActorTests.cs b/tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/DataConnectionActorTests.cs index f1547a44..0914ed78 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/DataConnectionActorTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/DataConnectionActorTests.cs @@ -463,6 +463,77 @@ public class DataConnectionActorTests : TestKit "sensor/temp", Arg.Any(), Arg.Any()); } + // ── 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 { ["Endpoint"] = "opc.tcp://primary:4840" }; + var backupConfig = new Dictionary { ["Endpoint"] = "opc.tcp://backup:4840" }; + var primaryAdapter = Substitute.For(); + var backupAdapter = Substitute.For(); + + // Primary connects once, then every reconnect fails so the disconnect drives failover. + var primaryConnectCount = 0; + primaryAdapter.ConnectAsync(Arg.Any>(), Arg.Any()) + .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(TaskCreationOptions.RunContinuationsAsynchronously); + primaryAdapter.SubscribeAsync(Arg.Any(), Arg.Any(), Arg.Any()) + .Returns(_ => subscribeGate.Task, _ => Task.FromResult("sub-primary-2")); + primaryAdapter.ReadAsync(Arg.Any(), Arg.Any()) + .Returns(new ReadResult(false, null, null)); + + _mockFactory.Create("OpcUa", Arg.Is>(d => d["Endpoint"] == "opc.tcp://backup:4840")) + .Returns(backupAdapter); + backupAdapter.ConnectAsync(Arg.Any>(), Arg.Any()) + .Returns(Task.CompletedTask); + backupAdapter.SubscribeAsync(Arg.Any(), Arg.Any(), Arg.Any()) + .Returns("sub-backup"); + backupAdapter.ReadAsync(Arg.Any(), Arg.Any()) + .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 DelayedSubscribeAsync()