feat(site-runtime): tag-subscribe in-flight guard — one outstanding request per connection until TagSubscribeInFlightCeilingMs (Lever 1)

FailedSubscribeResponse_IsRetried and LostSubscribeResponse_IsRetried_AndSuccessStopsRetrying now pin the in-flight ceiling to the retry interval: a lost reply is re-sent only once the outstanding request is presumed lost, by design.

Claude-Session: https://claude.ai/code/session_019ZxXT5sxZs5XrtdwYzJDPV

src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/InstanceActor.cs

@@ -178,14 +178,30 @@ 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
+    // Tag-subscription retry state (S4/UA6 + Lever 1 2026-08-28): tag paths grouped by
-    // (built once in SubscribeToDcl), the in-flight correlation-id → connection
+    // connection name (built once in SubscribeToDcl), the in-flight correlation-id →
-    // map, and a per-connection retry timer. Arming the timer on *send* closes both
+    // connection map, and per-connection retry state. Arming the watchdog on *send*
-    // the failed-response and the lost-response gaps with one mechanism; the timer
+    // closes both the failed-response and the lost-response gaps with one mechanism;
-    // is cancelled when a Success reply arrives.
+    // it is cancelled when a Success reply arrives. Lever 1 adds the in-flight guard
+    // (a tick while a young request is outstanding re-arms instead of re-sending) and
+    // exponential backoff (see NextTagSubscribeInterval) so a slow gateway is never
+    // handed a fresh N-tag subscribe on top of the one it is still processing.
     private readonly Dictionary<string, List<string>> _tagsByConnection = new();
     private readonly Dictionary<string, string> _pendingTagSubscribes = new();
-    private readonly Dictionary<string, ICancelable> _tagSubscribeRetryTimers = new();
+    private readonly Dictionary<string, TagSubscribeState> _tagSubscribeStates = new();
+
+    /// <summary>Per-connection tag-subscribe retry bookkeeping (Lever 1).</summary>
+    private sealed class TagSubscribeState
+    {
+        /// <summary>Correlation id of the outstanding request, null when none is in flight.</summary>
+        public string? InFlightCorrelationId;
+        /// <summary>When the outstanding request was sent (monotonic, for the ceiling check).</summary>
+        public long InFlightSentTicks;
+        /// <summary>Interval the NEXT retry will use; starts at the floor, grows on failure.</summary>
+        public TimeSpan Interval;
+        /// <summary>The armed watchdog for this connection.</summary>
+        public ICancelable? Timer;
+    }
 
     /// <summary>
     /// Initializes the instance actor with its configuration and dependencies.
@@ -388,10 +404,10 @@ public class InstanceActor : ReceiveActor
     /// <inheritdoc />
     protected override void PostStop()
     {
-        // Cancel any armed tag-subscribe retry timers so they cannot fire after stop.
+        // Cancel any armed tag-subscribe watchdogs so they cannot fire after stop.
-        foreach (var timer in _tagSubscribeRetryTimers.Values)
+        foreach (var state in _tagSubscribeStates.Values)
-            timer.Cancel();
+            state.Timer?.Cancel();
-        _tagSubscribeRetryTimers.Clear();
+        _tagSubscribeStates.Clear();
 
         // Final flush of any coalesced static-override writes so a graceful stop does not lose
         // the last <= flush-interval of writes (P4 pattern, mirrors NativeAlarmActor.PostStop).
@@ -1246,10 +1262,17 @@ public class InstanceActor : ReceiveActor
 
     /// <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
+    /// per-connection watchdog (S4/UA6). Arming on send means the retry fires whether
-    /// whether the response is a failure OR is lost entirely; a Success reply
+    /// the response is a failure OR is lost entirely; a Success reply cancels it.
-    /// cancels it. A superseded in-flight correlation id is overwritten so only the
+    /// <para>
-    /// latest attempt's reply is honoured.
+    /// Lever 1 (2026-08-28): if a request for this connection is already in flight and
+    /// younger than <see cref="SiteRuntimeOptions.TagSubscribeInFlightCeilingMs"/>, no
+    /// new request is issued — the watchdog is simply re-armed at the current interval
+    /// and the outstanding request keeps its correlation id. This is the guard that
+    /// stops a slow gateway from receiving one new N-tag subscribe per interval. Past
+    /// the ceiling the reply is presumed lost: the old id is dropped (its late reply, if
+    /// any, is ignored as superseded), the interval backs off, and a fresh request goes out.
+    /// </para>
     /// </summary>
     private void SendTagSubscribe(string connectionName)
     {
@@ -1257,23 +1280,73 @@ public class InstanceActor : ReceiveActor
         if (!_tagsByConnection.TryGetValue(connectionName, out var tagPaths) || tagPaths.Count == 0)
             return;
 
+        var state = GetTagSubscribeState(connectionName);
+        var now = Environment.TickCount64;
+
+        if (state.InFlightCorrelationId != null)
+        {
+            var ageMs = now - state.InFlightSentTicks;
+            if (ageMs < _options.TagSubscribeInFlightCeilingMs)
+            {
+                _logger.LogDebug(
+                    "Instance {Instance} tag subscribe on connection {Connection} still in flight ({AgeMs}ms, corr={Corr}) — not re-sending",
+                    _instanceUniqueName, connectionName, ageMs, state.InFlightCorrelationId);
+                ArmTagSubscribeWatchdog(connectionName, state);
+                return;
+            }
+
+            _logger.LogWarning(
+                "Instance {Instance} tag subscribe on connection {Connection} unanswered for {AgeMs}ms (corr={Corr}) — presuming lost, re-sending",
+                _instanceUniqueName, connectionName, ageMs, state.InFlightCorrelationId);
+            _pendingTagSubscribes.Remove(state.InFlightCorrelationId);
+            BackOffTagSubscribe(state);
+        }
+
         var correlationId = Guid.NewGuid().ToString("N");
         _pendingTagSubscribes[correlationId] = connectionName;
+        state.InFlightCorrelationId = correlationId;
+        state.InFlightSentTicks = now;
 
         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})",
+            "Instance {Instance} subscribing to {Count} tags on connection {Connection} (corr={Corr}, next retry in {Interval})",
-            _instanceUniqueName, tagPaths.Count, connectionName, correlationId);
+            _instanceUniqueName, tagPaths.Count, connectionName, correlationId, state.Interval);
 
-        if (_tagSubscribeRetryTimers.Remove(connectionName, out var existing))
+        ArmTagSubscribeWatchdog(connectionName, state);
-            existing.Cancel();
-        _tagSubscribeRetryTimers[connectionName] = Context.System.Scheduler.ScheduleTellOnceCancelable(
-            TimeSpan.FromMilliseconds(_options.TagSubscribeRetryIntervalMs), Self,
-            new RetryTagSubscribe(connectionName), Self);
     }
 
+    private TagSubscribeState GetTagSubscribeState(string connectionName)
+    {
+        if (!_tagSubscribeStates.TryGetValue(connectionName, out var state))
+        {
+            state = new TagSubscribeState { Interval = TimeSpan.FromMilliseconds(_options.TagSubscribeRetryIntervalMs) };
+            _tagSubscribeStates[connectionName] = state;
+        }
+        return state;
+    }
+
+    /// <summary>(Re)arms the per-connection watchdog at the state's current interval.</summary>
+    private void ArmTagSubscribeWatchdog(string connectionName, TagSubscribeState state)
+    {
+        state.Timer?.Cancel();
+        state.Timer = Context.System.Scheduler.ScheduleTellOnceCancelable(
+            state.Interval, Self, new RetryTagSubscribe(connectionName), Self);
+    }
+
+    /// <summary>Advances the backoff after a failed or presumed-lost attempt.</summary>
+    private void BackOffTagSubscribe(TagSubscribeState state) =>
+        state.Interval = NextTagSubscribeInterval(
+            state.Interval,
+            TimeSpan.FromMilliseconds(_options.TagSubscribeRetryIntervalMs),
+            TimeSpan.FromMilliseconds(_options.TagSubscribeRetryMaxIntervalMs),
+            Random.Shared.NextDouble());
+
+    /// <summary>Resets the backoff to its floor (Success, or connection back to Good).</summary>
+    private void ResetTagSubscribeBackoff(TagSubscribeState state) =>
+        state.Interval = TimeSpan.FromMilliseconds(_options.TagSubscribeRetryIntervalMs);
+
     /// <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
@@ -1286,8 +1359,14 @@ public class InstanceActor : ReceiveActor
 
         if (response.Success)
         {
-            if (_tagSubscribeRetryTimers.Remove(connectionName, out var timer))
+            var state = GetTagSubscribeState(connectionName);
-                timer.Cancel();
+            if (state.InFlightCorrelationId == response.CorrelationId)
+            {
+                state.Timer?.Cancel();
+                state.Timer = null;
+                state.InFlightCorrelationId = null;
+                ResetTagSubscribeBackoff(state);
+            }
             _logger.LogDebug(
                 "Instance {Instance} tag subscription confirmed on connection {Connection}",
                 _instanceUniqueName, connectionName);

tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Actors/InstanceActorTests.cs

@@ -1290,7 +1290,7 @@ public class InstanceActorTests : TestKit, IDisposable
     [Fact]
     public void FailedSubscribeResponse_IsRetried()
     {
-        var options = new SiteRuntimeOptions { TagSubscribeRetryIntervalMs = 200 };
+        var options = new SiteRuntimeOptions { TagSubscribeRetryIntervalMs = 200, TagSubscribeInFlightCeilingMs = 200 };
         var (instance, dclProbe) = CreateInstanceWithDataSourcedAttribute(options);
 
         var first = dclProbe.ExpectMsg<SubscribeTagsRequest>();
@@ -1305,7 +1305,8 @@ public class InstanceActorTests : TestKit, IDisposable
     [Fact]
     public void LostSubscribeResponse_IsRetried_AndSuccessStopsRetrying()
     {
-        var options = new SiteRuntimeOptions { TagSubscribeRetryIntervalMs = 200 };
+        // Lever 1: a lost reply is re-sent only once the outstanding request is presumed lost (ceiling), by design.
+        var options = new SiteRuntimeOptions { TagSubscribeRetryIntervalMs = 200, TagSubscribeInFlightCeilingMs = 200 };
         var (instance, dclProbe) = CreateInstanceWithDataSourcedAttribute(options);
 
         dclProbe.ExpectMsg<SubscribeTagsRequest>();
@@ -1316,4 +1317,68 @@ public class InstanceActorTests : TestKit, IDisposable
             true, null, DateTimeOffset.UtcNow), dclProbe.Ref);
         dclProbe.ExpectNoMsg(TimeSpan.FromMilliseconds(600)); // retry timer cancelled
     }
+
+    // --- Lever 1 (2026-08-28): in-flight guard ---
+
+    [Fact]
+    public void SlowSubscribe_DoesNotStackDuplicates_WhileInFlight()
+    {
+        // Retry ticks every 200 ms, but an unanswered request blocks a duplicate for 2 s.
+        var options = new SiteRuntimeOptions
+        {
+            TagSubscribeRetryIntervalMs = 200,
+            TagSubscribeRetryMaxIntervalMs = 200,
+            TagSubscribeInFlightCeilingMs = 2000,
+        };
+        var (_, dclProbe) = CreateInstanceWithDataSourcedAttribute(options);
+
+        dclProbe.ExpectMsg<SubscribeTagsRequest>();
+        // Pre-Lever-1 this window would carry ~4 more requests (one per 200 ms tick).
+        dclProbe.ExpectNoMsg(TimeSpan.FromMilliseconds(1000));
+    }
+
+    [Fact]
+    public void SlowSubscribe_ReissuesAfterInFlightCeiling_AndOnlyOnce()
+    {
+        var options = new SiteRuntimeOptions
+        {
+            TagSubscribeRetryIntervalMs = 200,
+            TagSubscribeRetryMaxIntervalMs = 200,
+            TagSubscribeInFlightCeilingMs = 600,
+        };
+        var (_, dclProbe) = CreateInstanceWithDataSourcedAttribute(options);
+
+        var first = dclProbe.ExpectMsg<SubscribeTagsRequest>();
+        var second = dclProbe.ExpectMsg<SubscribeTagsRequest>(TimeSpan.FromSeconds(3)); // after ceiling
+        Assert.NotEqual(first.CorrelationId, second.CorrelationId);
+        Assert.Equal(first.TagPaths, second.TagPaths);
+        // The new request is itself guarded — no third one inside its ceiling window.
+        dclProbe.ExpectNoMsg(TimeSpan.FromMilliseconds(400));
+    }
+
+    [Fact]
+    public void ReplyToSupersededRequest_IsIgnored_AndLatestStillGoverns()
+    {
+        var options = new SiteRuntimeOptions
+        {
+            TagSubscribeRetryIntervalMs = 200,
+            TagSubscribeRetryMaxIntervalMs = 200,
+            TagSubscribeInFlightCeilingMs = 300,
+        };
+        var (instance, dclProbe) = CreateInstanceWithDataSourcedAttribute(options);
+
+        var first = dclProbe.ExpectMsg<SubscribeTagsRequest>();
+        var second = dclProbe.ExpectMsg<SubscribeTagsRequest>(TimeSpan.FromSeconds(3));
+
+        // A late Success for the FIRST id must not cancel the watchdog guarding the SECOND.
+        instance.Tell(new SubscribeTagsResponse(first.CorrelationId, first.InstanceUniqueName,
+            true, null, DateTimeOffset.UtcNow), dclProbe.Ref);
+        var third = dclProbe.ExpectMsg<SubscribeTagsRequest>(TimeSpan.FromSeconds(3)); // second eventually re-sent
+        Assert.NotEqual(second.CorrelationId, third.CorrelationId);
+
+        // Success for the live id stops everything.
+        instance.Tell(new SubscribeTagsResponse(third.CorrelationId, third.InstanceUniqueName,
+            true, null, DateTimeOffset.UtcNow), dclProbe.Ref);
+        dclProbe.ExpectNoMsg(TimeSpan.FromMilliseconds(800));
+    }
 }