diff --git a/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/InstanceActor.cs b/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/InstanceActor.cs index d73356ca..a36ded6c 100644 --- a/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/InstanceActor.cs +++ b/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> _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. + // Tag-subscription retry state (S4/UA6 + Lever 1 2026-08-28): tag paths grouped by + // connection name (built once in SubscribeToDcl), the in-flight correlation-id → + // connection map, and per-connection retry state. Arming the watchdog on *send* + // closes both the failed-response and the lost-response gaps with one mechanism; + // 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> _tagsByConnection = new(); private readonly Dictionary _pendingTagSubscribes = new(); - private readonly Dictionary _tagSubscribeRetryTimers = new(); + private readonly Dictionary _tagSubscribeStates = new(); + + /// Per-connection tag-subscribe retry bookkeeping (Lever 1). + private sealed class TagSubscribeState + { + /// Correlation id of the outstanding request, null when none is in flight. + public string? InFlightCorrelationId; + /// When the outstanding request was sent (monotonic, for the ceiling check). + public long InFlightSentTicks; + /// Interval the NEXT retry will use; starts at the floor, grows on failure. + public TimeSpan Interval; + /// The armed watchdog for this connection. + public ICancelable? Timer; + } /// /// Initializes the instance actor with its configuration and dependencies. @@ -388,10 +404,10 @@ 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(); + // Cancel any armed tag-subscribe watchdogs so they cannot fire after stop. + foreach (var state in _tagSubscribeStates.Values) + state.Timer?.Cancel(); + _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 /// /// 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. + /// per-connection watchdog (S4/UA6). Arming on send means the retry fires whether + /// the response is a failure OR is lost entirely; a Success reply cancels it. + /// + /// Lever 1 (2026-08-28): if a request for this connection is already in flight and + /// younger than , 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. + /// /// 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})", - _instanceUniqueName, tagPaths.Count, connectionName, correlationId); + "Instance {Instance} subscribing to {Count} tags on connection {Connection} (corr={Corr}, next retry in {Interval})", + _instanceUniqueName, tagPaths.Count, connectionName, correlationId, state.Interval); - if (_tagSubscribeRetryTimers.Remove(connectionName, out var existing)) - existing.Cancel(); - _tagSubscribeRetryTimers[connectionName] = Context.System.Scheduler.ScheduleTellOnceCancelable( - TimeSpan.FromMilliseconds(_options.TagSubscribeRetryIntervalMs), Self, - new RetryTagSubscribe(connectionName), Self); + ArmTagSubscribeWatchdog(connectionName, state); } + 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; + } + + /// (Re)arms the per-connection watchdog at the state's current interval. + private void ArmTagSubscribeWatchdog(string connectionName, TagSubscribeState state) + { + state.Timer?.Cancel(); + state.Timer = Context.System.Scheduler.ScheduleTellOnceCancelable( + state.Interval, Self, new RetryTagSubscribe(connectionName), Self); + } + + /// Advances the backoff after a failed or presumed-lost attempt. + private void BackOffTagSubscribe(TagSubscribeState state) => + state.Interval = NextTagSubscribeInterval( + state.Interval, + TimeSpan.FromMilliseconds(_options.TagSubscribeRetryIntervalMs), + TimeSpan.FromMilliseconds(_options.TagSubscribeRetryMaxIntervalMs), + Random.Shared.NextDouble()); + + /// Resets the backoff to its floor (Success, or connection back to Good). + private void ResetTagSubscribeBackoff(TagSubscribeState state) => + state.Interval = TimeSpan.FromMilliseconds(_options.TagSubscribeRetryIntervalMs); + /// /// 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)) - timer.Cancel(); + var state = GetTagSubscribeState(connectionName); + 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); 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 fad294d4..2a970815 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Actors/InstanceActorTests.cs +++ b/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(); @@ -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(); @@ -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(); + // 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(); + var second = dclProbe.ExpectMsg(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(); + var second = dclProbe.ExpectMsg(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(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)); + } }