From b6b401c879e84bbc94000b4d44d37786c4b3c3d7 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Fri, 28 Aug 2026 14:36:22 -0400 Subject: [PATCH] =?UTF-8?q?feat(site-runtime):=20tag-subscribe=20failure?= =?UTF-8?q?=20reply=20frees=20the=20in-flight=20slot,=20backs=20off=20(?= =?UTF-8?q?=C3=972=20=C2=B120%,=20capped)=20and=20Success=20resets=20to=20?= =?UTF-8?q?floor=20(Lever=201)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Also from the Task 3 review: FailedSubscribeResponse_IsRetried guards the failure path again (large ceiling); HandleSubscribeTagsResponse reads state instead of get-or-create; SubscribeToDcl prunes stale retry state on re-entry; doc comments refreshed. Claude-Session: https://claude.ai/code/session_019ZxXT5sxZs5XrtdwYzJDPV --- .../Actors/InstanceActor.cs | 44 ++++++++++--- .../Actors/InstanceActorTests.cs | 61 ++++++++++++++++++- 2 files changed, 95 insertions(+), 10 deletions(-) diff --git a/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/InstanceActor.cs b/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/InstanceActor.cs index a36ded6c..c1280e22 100644 --- a/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/InstanceActor.cs +++ b/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/InstanceActor.cs @@ -186,6 +186,8 @@ public class InstanceActor : ReceiveActor // (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. + // A re-entry into SubscribeToDcl rebuilds all three together: the retry state is + // pruned (timers cancelled, pending rows dropped) before the new grouping is built. private readonly Dictionary> _tagsByConnection = new(); private readonly Dictionary _pendingTagSubscribes = new(); private readonly Dictionary _tagSubscribeStates = new(); @@ -1206,6 +1208,16 @@ public class InstanceActor : ReceiveActor { if (_dclManager == null || _configuration == null) return; + // Lever 1: prune retry state from a previous build — cancels any armed watchdog and + // drops its pending row so a re-entry can never strand an in-flight slot. + foreach (var state in _tagSubscribeStates.Values) + { + state.Timer?.Cancel(); + if (state.InFlightCorrelationId != null) + _pendingTagSubscribes.Remove(state.InFlightCorrelationId); + } + _tagSubscribeStates.Clear(); + // Group attributes by their bound connection name into the persistent // _tagsByConnection field so a retry can re-send the exact same request. _tagsByConnection.Clear(); @@ -1348,18 +1360,27 @@ public class InstanceActor : ReceiveActor 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 - /// to re-send. A reply for an unknown/superseded correlation id is ignored. + /// Handles the DCL's tag-subscribe response. A reply whose correlation id is not the + /// pending one (superseded at the in-flight ceiling, or a duplicate) is ignored. On + /// Success the connection's watchdog is cancelled, the in-flight slot cleared and the + /// backoff reset to the floor. On failure (Lever 1, 2026-08-28) the slot is freed, the + /// backoff advanced (×2 ±20 % jitter, capped) and the watchdog re-armed at the new + /// interval, so the next attempt waits the backed-off interval — not the in-flight + /// ceiling — and duplicates are still impossible. /// private void HandleSubscribeTagsResponse(SubscribeTagsResponse response) { if (!_pendingTagSubscribes.Remove(response.CorrelationId, out var connectionName)) return; // stale/superseded reply — ignore + // Read, never create: the id was minted by SendTagSubscribe, which creates the + // state, so a missing entry means the retry state was pruned (SubscribeToDcl + // re-entry) and this reply no longer governs anything. + if (!_tagSubscribeStates.TryGetValue(connectionName, out var state)) + return; + if (response.Success) { - var state = GetTagSubscribeState(connectionName); if (state.InFlightCorrelationId == response.CorrelationId) { state.Timer?.Cancel(); @@ -1373,14 +1394,21 @@ public class InstanceActor : ReceiveActor } else { + if (state.InFlightCorrelationId == response.CorrelationId) + { + // A definite failure: the slot is free, back off, and re-arm at the NEW + // interval so the backoff is honoured on this very retry, not the one after. + state.InFlightCorrelationId = null; + BackOffTagSubscribe(state); + ArmTagSubscribeWatchdog(connectionName, state); + } _logger.LogWarning( - "Instance {Instance} tag subscribe FAILED on connection {Connection}: {Error} — will retry", - _instanceUniqueName, connectionName, response.ErrorMessage); + "Instance {Instance} tag subscribe FAILED on connection {Connection}: {Error} — will retry in {Interval}", + _instanceUniqueName, connectionName, response.ErrorMessage, state.Interval); _ = _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. + $"Tag subscribe failed on connection '{connectionName}': {response.ErrorMessage} — will retry in {state.Interval}"); } } 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 2a970815..9bc1508d 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, TagSubscribeInFlightCeilingMs = 200 }; + var options = new SiteRuntimeOptions { TagSubscribeRetryIntervalMs = 200, TagSubscribeInFlightCeilingMs = 60_000 }; var (instance, dclProbe) = CreateInstanceWithDataSourcedAttribute(options); var first = dclProbe.ExpectMsg(); @@ -1370,7 +1370,8 @@ public class InstanceActorTests : TestKit, IDisposable 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. + // A late Success for the FIRST id is dropped at the pending-map check (its id was + // removed at the ceiling); the watchdog guarding the SECOND is untouched. 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 @@ -1381,4 +1382,60 @@ public class InstanceActorTests : TestKit, IDisposable true, null, DateTimeOffset.UtcNow), dclProbe.Ref); dclProbe.ExpectNoMsg(TimeSpan.FromMilliseconds(800)); } + + // --- Lever 1 (2026-08-28): backoff on failure, reset on success --- + + [Fact] + public void RepeatedFailures_BackOff_AndSuccessResetsToFloor() + { + // Floor 200 ms, max 800 ms. Jitter is ±20 % so bound the assertions, don't pin them. + var options = new SiteRuntimeOptions + { + TagSubscribeRetryIntervalMs = 200, + TagSubscribeRetryMaxIntervalMs = 800, + TagSubscribeInFlightCeilingMs = 60_000, // guard never expires; only replies drive retries + }; + var (instance, dclProbe) = CreateInstanceWithDataSourcedAttribute(options); + + var req = dclProbe.ExpectMsg(); + var gaps = new List(); + for (var i = 0; i < 3; i++) + { + instance.Tell(new SubscribeTagsResponse(req.CorrelationId, req.InstanceUniqueName, + false, "gateway busy", DateTimeOffset.UtcNow), dclProbe.Ref); + var sw = System.Diagnostics.Stopwatch.StartNew(); + req = dclProbe.ExpectMsg(TimeSpan.FromSeconds(5)); + gaps.Add(sw.Elapsed); + } + + // Expected nominal gaps: 400, 800(cap), 800(cap) — each ±20 % jitter and scheduler slack. + Assert.InRange(gaps[0].TotalMilliseconds, 300, 560); + Assert.InRange(gaps[1].TotalMilliseconds, 600, 1100); + Assert.InRange(gaps[2].TotalMilliseconds, 600, 1100); + + // Success resets: no further request follows a confirmed subscription. + instance.Tell(new SubscribeTagsResponse(req.CorrelationId, req.InstanceUniqueName, + true, null, DateTimeOffset.UtcNow), dclProbe.Ref); + dclProbe.ExpectNoMsg(TimeSpan.FromMilliseconds(1200)); + } + + [Fact] + public void FailureReply_ReissuesEvenThoughGuardHasNotExpired() + { + var options = new SiteRuntimeOptions + { + TagSubscribeRetryIntervalMs = 200, + TagSubscribeRetryMaxIntervalMs = 200, + TagSubscribeInFlightCeilingMs = 60_000, + }; + 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); + + // A definite failure clears the in-flight slot, so the watchdog re-sends. + var retry = dclProbe.ExpectMsg(TimeSpan.FromSeconds(3)); + Assert.NotEqual(first.CorrelationId, retry.CorrelationId); + } }