fix(site-runtime): InstanceActor retries failed/lost tag subscriptions per connection (S4/UA6) — closes the unknown-connection ordering race

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<string, List<string>> _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<string, List<string>> _tagsByConnection = new();
+    private readonly Dictionary<string, string> _pendingTagSubscribes = new();
+    private readonly Dictionary<string, ICancelable> _tagSubscribeRetryTimers = new();
+
     /// <summary>
     /// Initializes the instance actor with its configuration and dependencies.
     /// </summary>
@@ -204,7 +213,8 @@ public class InstanceActor : ReceiveActor
 
         // Handle tag value updates from DCL — convert to AttributeValueChanged
         Receive<TagValueUpdate>(HandleTagValueUpdate);
-        Receive<SubscribeTagsResponse>(_ => { }); // Ack from DCL subscribe — no action needed
+        Receive<SubscribeTagsResponse>(HandleSubscribeTagsResponse); // S4/UA6: real retry-aware handler
+        Receive<RetryTagSubscribe>(msg => SendTagSubscribe(msg.ConnectionName));
         Receive<ConnectionQualityChanged>(HandleConnectionQualityChanged);
 
         // Handle alarm state changes from Alarm Actors (Tell pattern)
@@ -257,6 +267,11 @@ public class InstanceActor : ReceiveActor
     /// <inheritdoc />
     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<string, List<string>>();
+        // 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<string>();
-                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);
+    }
+
+    /// <summary>
+    /// 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.
+    /// </summary>
+    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(
-                Guid.NewGuid().ToString("N"),
-                _instanceUniqueName,
-                connectionName,
-                tagPaths,
-                DateTimeOffset.UtcNow);
+            correlationId, _instanceUniqueName, connectionName, tagPaths, DateTimeOffset.UtcNow);
         _dclManager.Tell(request, Self);
         _logger.LogInformation(
-                "Instance {Instance} subscribed to {Count} tags on connection {Connection}",
-                _instanceUniqueName, tagPaths.Count, connectionName);
+            "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);
+    }
+
+    /// <summary>
+    /// 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.
+    /// </summary>
+    private void HandleSubscribeTagsResponse(SubscribeTagsResponse response)
+    {
+        if (!_pendingTagSubscribes.Remove(response.CorrelationId, out var connectionName))
+            return; // stale/superseded reply — ignore
+
+        if (response.Success)
+        {
+            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<ISiteEventLogger>()?.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);
+
+    /// <summary>
+    /// Self-scheduled tick (S4/UA6): re-send the tag-subscribe request for one
+    /// connection whose previous attempt failed or whose response was lost.
+    /// </summary>
+    private sealed record RetryTagSubscribe(string ConnectionName);
 }

src/ZB.MOM.WW.ScadaBridge.SiteRuntime/SiteRuntimeOptions.cs

@@ -66,6 +66,14 @@ public class SiteRuntimeOptions
     /// </summary>
     public int ConfigFetchRetryCount { get; set; } = 3;
 
+    /// <summary>
+    /// 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.
+    /// </summary>
+    public int TagSubscribeRetryIntervalMs { get; set; } = 5000;
+
     /// <summary>
     /// Grace period (ms) after a script's execution timeout elapses before the
     /// stuck-script watchdog declares the script's dedicated thread blocked and

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) ──
+
+    /// <summary>
+    /// 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.
+    /// </summary>
+    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<InstanceActor>.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<SubscribeTagsRequest>();
+        instance.Tell(new SubscribeTagsResponse(first.CorrelationId, first.InstanceUniqueName,
+            false, "Unknown connection: conn-1", DateTimeOffset.UtcNow), dclProbe.Ref);
+
+        var retry = dclProbe.ExpectMsg<SubscribeTagsRequest>(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<SubscribeTagsRequest>();
+        // No reply at all (dead-lettered response) -> must re-send.
+        var retry = dclProbe.ExpectMsg<SubscribeTagsRequest>(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
+    }
 }