From feb6f34e907d3e918599f254fd73b4ea86de93bc Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 9 Jul 2026 00:03:44 -0400 Subject: [PATCH] =?UTF-8?q?fix(site-runtime):=20InstanceActor=20retries=20?= =?UTF-8?q?failed/lost=20tag=20subscriptions=20per=20connection=20(S4/UA6)?= =?UTF-8?q?=20=E2=80=94=20closes=20the=20unknown-connection=20ordering=20r?= =?UTF-8?q?ace?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../Actors/InstanceActor.cs | 106 +++++++++++++++--- .../SiteRuntimeOptions.cs | 8 ++ .../Actors/InstanceActorTests.cs | 65 +++++++++++ 3 files changed, 162 insertions(+), 17 deletions(-) diff --git a/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/InstanceActor.cs b/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/InstanceActor.cs index 0e79f3e2..0fdf6717 100644 --- a/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/InstanceActor.cs +++ b/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/InstanceActor.cs @@ -96,6 +96,15 @@ public class InstanceActor : ReceiveActor // fan out to all of them — not just the last one registered. private readonly Dictionary> _tagPathToAttributes = new(); + // Tag-subscription retry state (S4/UA6): tag paths grouped by connection name + // (built once in SubscribeToDcl), the in-flight correlation-id → connection + // map, and a per-connection retry timer. Arming the timer on *send* closes both + // the failed-response and the lost-response gaps with one mechanism; the timer + // is cancelled when a Success reply arrives. + private readonly Dictionary> _tagsByConnection = new(); + private readonly Dictionary _pendingTagSubscribes = new(); + private readonly Dictionary _tagSubscribeRetryTimers = new(); + /// /// Initializes the instance actor with its configuration and dependencies. /// @@ -204,7 +213,8 @@ public class InstanceActor : ReceiveActor // Handle tag value updates from DCL — convert to AttributeValueChanged Receive(HandleTagValueUpdate); - Receive(_ => { }); // Ack from DCL subscribe — no action needed + Receive(HandleSubscribeTagsResponse); // S4/UA6: real retry-aware handler + Receive(msg => SendTagSubscribe(msg.ConnectionName)); Receive(HandleConnectionQualityChanged); // Handle alarm state changes from Alarm Actors (Tell pattern) @@ -257,6 +267,11 @@ public class InstanceActor : ReceiveActor /// protected override void PostStop() { + // Cancel any armed tag-subscribe retry timers so they cannot fire after stop. + foreach (var timer in _tagSubscribeRetryTimers.Values) + timer.Cancel(); + _tagSubscribeRetryTimers.Clear(); + // Operational `instance_lifecycle` event — instance stopped. An // instance stops on disable, delete, redeployment, and graceful shutdown; // this single point covers them all. @@ -963,8 +978,9 @@ public class InstanceActor : ReceiveActor { if (_dclManager == null || _configuration == null) return; - // Group attributes by their bound connection name - var byConnection = new Dictionary>(); + // Group attributes by their bound connection name into the persistent + // _tagsByConnection field so a retry can re-send the exact same request. + _tagsByConnection.Clear(); foreach (var attr in _configuration.Attributes) { if (string.IsNullOrEmpty(attr.DataSourceReference) || @@ -980,10 +996,10 @@ public class InstanceActor : ReceiveActor } attrs.Add(attr.CanonicalName); - if (!byConnection.TryGetValue(attr.BoundDataConnectionName, out var connTags)) + if (!_tagsByConnection.TryGetValue(attr.BoundDataConnectionName, out var connTags)) { connTags = new List(); - byConnection[attr.BoundDataConnectionName] = connTags; + _tagsByConnection[attr.BoundDataConnectionName] = connTags; } // Subscribe each distinct tag path once per connection — a tag shared // by several attributes still needs only one DCL subscription. @@ -991,19 +1007,69 @@ public class InstanceActor : ReceiveActor connTags.Add(attr.DataSourceReference); } - // Send subscription requests to DCL for each connection - foreach (var (connectionName, tagPaths) in byConnection) + // Send subscription requests to DCL for each connection (retry-armed). + foreach (var connectionName in _tagsByConnection.Keys) + SendTagSubscribe(connectionName); + } + + /// + /// Sends (or re-sends) the tag-subscribe request for one connection and arms a + /// per-connection retry timer (S4/UA6). Arming on send means the retry fires + /// whether the response is a failure OR is lost entirely; a Success reply + /// cancels it. A superseded in-flight correlation id is overwritten so only the + /// latest attempt's reply is honoured. + /// + private void SendTagSubscribe(string connectionName) + { + if (_dclManager == null) return; + if (!_tagsByConnection.TryGetValue(connectionName, out var tagPaths) || tagPaths.Count == 0) + return; + + var correlationId = Guid.NewGuid().ToString("N"); + _pendingTagSubscribes[correlationId] = connectionName; + + var request = new SubscribeTagsRequest( + correlationId, _instanceUniqueName, connectionName, tagPaths, DateTimeOffset.UtcNow); + _dclManager.Tell(request, Self); + _logger.LogInformation( + "Instance {Instance} subscribing to {Count} tags on connection {Connection} (corr={Corr})", + _instanceUniqueName, tagPaths.Count, connectionName, correlationId); + + if (_tagSubscribeRetryTimers.Remove(connectionName, out var existing)) + existing.Cancel(); + _tagSubscribeRetryTimers[connectionName] = Context.System.Scheduler.ScheduleTellOnceCancelable( + TimeSpan.FromMilliseconds(_options.TagSubscribeRetryIntervalMs), Self, + new RetryTagSubscribe(connectionName), Self); + } + + /// + /// Handles the DCL's tag-subscribe response: on Success cancels the connection's + /// retry timer; on failure logs + emits a site event and leaves the armed timer + /// to re-send. A reply for an unknown/superseded correlation id is ignored. + /// + private void HandleSubscribeTagsResponse(SubscribeTagsResponse response) + { + if (!_pendingTagSubscribes.Remove(response.CorrelationId, out var connectionName)) + return; // stale/superseded reply — ignore + + if (response.Success) { - var request = new SubscribeTagsRequest( - Guid.NewGuid().ToString("N"), - _instanceUniqueName, - connectionName, - tagPaths, - DateTimeOffset.UtcNow); - _dclManager.Tell(request, Self); - _logger.LogInformation( - "Instance {Instance} subscribed to {Count} tags on connection {Connection}", - _instanceUniqueName, tagPaths.Count, connectionName); + if (_tagSubscribeRetryTimers.Remove(connectionName, out var timer)) + timer.Cancel(); + _logger.LogDebug( + "Instance {Instance} tag subscription confirmed on connection {Connection}", + _instanceUniqueName, connectionName); + } + else + { + _logger.LogWarning( + "Instance {Instance} tag subscribe FAILED on connection {Connection}: {Error} — will retry", + _instanceUniqueName, connectionName, response.ErrorMessage); + _ = _serviceProvider?.GetService()?.LogEventAsync( + "instance_lifecycle", "Warning", _instanceUniqueName, + $"InstanceActor:{_instanceUniqueName}", + $"Tag subscribe failed on connection '{connectionName}': {response.ErrorMessage} — will retry"); + // Leave the armed retry timer to re-send. } } @@ -1571,4 +1637,10 @@ public class InstanceActor : ReceiveActor IActorRef Replyer, ICancelable Timeout, bool RequireGoodQuality); + + /// + /// Self-scheduled tick (S4/UA6): re-send the tag-subscribe request for one + /// connection whose previous attempt failed or whose response was lost. + /// + private sealed record RetryTagSubscribe(string ConnectionName); } diff --git a/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/SiteRuntimeOptions.cs b/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/SiteRuntimeOptions.cs index 66c3bf98..a05e65d2 100644 --- a/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/SiteRuntimeOptions.cs +++ b/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/SiteRuntimeOptions.cs @@ -66,6 +66,14 @@ public class SiteRuntimeOptions /// public int ConfigFetchRetryCount { get; set; } = 3; + /// + /// Fixed interval (ms) at which an Instance Actor re-sends a tag-subscribe + /// request that either failed or whose response was lost (S4/UA6). The retry is + /// armed on every send and cancelled on the first Success reply, so it closes + /// both the failed-response and the lost-response gaps. Default: 5000ms. + /// + public int TagSubscribeRetryIntervalMs { get; set; } = 5000; + /// /// Grace period (ms) after a script's execution timeout elapses before the /// stuck-script watchdog declares the script's dedicated thread blocked and diff --git a/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Actors/InstanceActorTests.cs b/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Actors/InstanceActorTests.cs index 72dbb49c..bc8f3c31 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Actors/InstanceActorTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Actors/InstanceActorTests.cs @@ -1171,4 +1171,69 @@ public class InstanceActorTests : TestKit, IDisposable Assert.Single(overrides); Assert.Equal("[\"a\"", overrides["Counts"]); } + + // ── S4/UA6: tag-subscription retry (failed AND lost responses) ── + + /// + /// Builds an Instance Actor with a single data-sourced attribute and the given + /// options (so the retry interval is short), returning it plus the DCL probe. + /// The initial SubscribeTagsRequest is left in the probe's queue for the caller. + /// + private (IActorRef Instance, TestProbe Dcl) CreateInstanceWithDataSourcedAttribute(SiteRuntimeOptions options) + { + var config = new FlattenedConfiguration + { + InstanceUniqueName = "Sub-1", + Attributes = + [ + new ResolvedAttribute + { + CanonicalName = "Motor.Temp", Value = "0", DataType = "Int", + DataSourceReference = "ns=2;s=Motor.Temp", BoundDataConnectionName = "conn-1" + } + ] + }; + var dcl = CreateTestProbe(); + var actor = ActorOf(Props.Create(() => new InstanceActor( + "Sub-1", + JsonSerializer.Serialize(config), + _storage, + _compilationService, + _sharedScriptLibrary, + null, + options, + NullLogger.Instance, + dcl.Ref))); + return (actor, dcl); + } + + [Fact] + public void FailedSubscribeResponse_IsRetried() + { + var options = new SiteRuntimeOptions { TagSubscribeRetryIntervalMs = 200 }; + var (instance, dclProbe) = CreateInstanceWithDataSourcedAttribute(options); + + var first = dclProbe.ExpectMsg(); + instance.Tell(new SubscribeTagsResponse(first.CorrelationId, first.InstanceUniqueName, + false, "Unknown connection: conn-1", DateTimeOffset.UtcNow), dclProbe.Ref); + + var retry = dclProbe.ExpectMsg(TimeSpan.FromSeconds(5)); // S4 core + Assert.Equal(first.ConnectionName, retry.ConnectionName); + Assert.Equal(first.TagPaths, retry.TagPaths); + } + + [Fact] + public void LostSubscribeResponse_IsRetried_AndSuccessStopsRetrying() + { + var options = new SiteRuntimeOptions { TagSubscribeRetryIntervalMs = 200 }; + var (instance, dclProbe) = CreateInstanceWithDataSourcedAttribute(options); + + dclProbe.ExpectMsg(); + // No reply at all (dead-lettered response) -> must re-send. + var retry = dclProbe.ExpectMsg(TimeSpan.FromSeconds(5)); + + instance.Tell(new SubscribeTagsResponse(retry.CorrelationId, retry.InstanceUniqueName, + true, null, DateTimeOffset.UtcNow), dclProbe.Ref); + dclProbe.ExpectNoMsg(TimeSpan.FromMilliseconds(600)); // retry timer cancelled + } }