ScadaBridge/docs/plans/2026-08-28-tag-subscribe-inflight-guard-backoff.md at 16cdf0adf1b38df0cfd2fda641fa49e34618a7bb

Files

T

Joseph Doherty e39cca093b docs(plans): Lever 1 implementation plan — tag-subscribe in-flight guard + exponential backoff in InstanceActor (8 tasks, TDD)

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

2026-08-28 14:18:01 -04:00

45 KiB

Raw Blame History

For Claude: REQUIRED SUB-SKILL: Use superpowers-extended-cc:executing-plans to implement this plan task-by-task.

Goal: Stop InstanceActor from stacking duplicate tag-subscribe requests while one is still outstanding, and replace its fixed 5 s retry with exponential backoff + jitter — Lever 1 of docs/plans/2026-08-28-mxgateway-subscribe-storm-prevention-design.md §3.

Architecture: All changes are inside InstanceActor.SendTagSubscribe / HandleSubscribeTagsResponse (SiteRuntime) plus two new SiteRuntimeOptions keys. The pure backoff step is a static method with the same shape as DataConnectionActor.NextTagResolutionInterval so it is unit-testable without an actor. Per-connection state grows from one dictionary (_tagSubscribeRetryTimers) to a small record holding the in-flight correlation id + send time and the current interval. No wire, proto, DB, or cross-repo change.

Tech Stack: .NET 10, Akka.NET (Context.System.Scheduler.ScheduleTellOnceCancelable), xunit + Akka.TestKit.Xunit2, ZB.MOM.WW.Configuration OptionsValidatorBase.

Background the implementer needs

Today (src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/InstanceActor.cs): SendTagSubscribe(connectionName) (:1234) mints a correlation id, records it in _pendingTagSubscribes[corr] = connectionName, Tells the DCL manager a SubscribeTagsRequest, then unconditionally (re)arms a per-connection RetryTagSubscribe self-tell after TagSubscribeRetryIntervalMs (:1250-1254). The retry handler (:314) just calls SendTagSubscribe again. HandleSubscribeTagsResponse (:1262) drops replies whose correlation id is not in _pendingTagSubscribes, cancels the timer on Success (:1269), and on failure logs/site-events and lets the timer re-send. PostStop (:389-394) cancels timers.
Defect: nothing checks whether a subscribe is already in flight, and the interval never grows. A subscribe that takes > 5 s (gateway under load) gets a second, third, … concurrent 244-tag subscribe every 5 s. See design doc §1.
Existing tests (tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Actors/InstanceActorTests.cs:1262-1322): CreateInstanceWithDataSourcedAttribute(options) builds an InstanceActor with one data-sourced attribute on conn-1 and a TestProbe standing in for the DCL manager; the initial SubscribeTagsRequest lands in the probe. FailedSubscribeResponse_IsRetried and LostSubscribeResponse_IsRetried_AndSuccessStopsRetrying use TagSubscribeRetryIntervalMs = 200. Both are amended in Task 3 (they pin TagSubscribeInFlightCeilingMs = 200) so their intent survives the guard.
Pattern to mirror: DataConnectionActor.NextTagResolutionInterval(current, floor, max) (src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Actors/DataConnectionActor.cs:~2422) — doubles, caps at max, and treats max < floor as "fixed at floor". Don't call it (cross-project, internal); copy the shape.
Semantics to preserve: a lost reply must still eventually trigger a re-send (that is what "arm on send" bought us). The in-flight guard therefore has a ceiling: once the outstanding request is older than TagSubscribeInFlightCeilingMs, a new one is allowed even though no reply came.
No connection-restore hook exists in InstanceActor — the DCL re-subscribes transparently on reconnect and the Instance Actor only sees ConnectionQualityChanged. The design's "reset backoff on restore" maps to: reset on Success (Task 4) and reset when ConnectionQualityChanged reports Good (Task 6).

Decision points resolved here (do not reopen):

Jitter: ±20 % uniform, applied in the pure step so it's testable; the actor passes Random.Shared.NextDouble().
Backoff advances only on a Failure reply or a ceiling-expired lost reply, never on the in-flight "skip" path (a skip just re-arms the watchdog at the current interval).
New options default: TagSubscribeRetryMaxIntervalMs = 60000, TagSubscribeInFlightCeilingMs = 45000.

Task 1: Options + validator (new keys)

Classification: small Estimated implement time: ~3 min Parallelizable with: Task 2

Files:

Modify: src/ZB.MOM.WW.ScadaBridge.SiteRuntime/SiteRuntimeOptions.cs:114-120
Modify: src/ZB.MOM.WW.ScadaBridge.SiteRuntime/SiteRuntimeOptionsValidator.cs:77-80
Test: tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/SiteRuntimeOptionsValidatorTests.cs

Step 1: Write the failing tests — append inside the class in SiteRuntimeOptionsValidatorTests.cs:

    // --- Lever 1 (2026-08-28): tag-subscribe backoff + in-flight ceiling ---

    [Fact]
    public void TagSubscribeRetryMaxIntervalMs_BelowFloor_IsRejected()
    {
        var result = Validate(new SiteRuntimeOptions
        {
            TagSubscribeRetryIntervalMs = 5000,
            TagSubscribeRetryMaxIntervalMs = 4999,
        });

        Assert.True(result.Failed);
        Assert.Contains("TagSubscribeRetryMaxIntervalMs", result.FailureMessage);
    }

    [Fact]
    public void TagSubscribeRetryMaxIntervalMs_EqualToFloor_IsValid()
    {
        var result = Validate(new SiteRuntimeOptions
        {
            TagSubscribeRetryIntervalMs = 5000,
            TagSubscribeRetryMaxIntervalMs = 5000,
        });
        Assert.True(result.Succeeded, result.FailureMessage);
    }

    [Fact]
    public void ZeroTagSubscribeInFlightCeilingMs_IsRejected()
    {
        var result = Validate(new SiteRuntimeOptions { TagSubscribeInFlightCeilingMs = 0 });

        Assert.True(result.Failed);
        Assert.Contains("TagSubscribeInFlightCeilingMs", result.FailureMessage);
    }

Step 2: Run to verify they fail

Run: dotnet test tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests --filter "FullyQualifiedName~SiteRuntimeOptionsValidatorTests" 2>&1 | tail -5 Expected: build error 'SiteRuntimeOptions' does not contain a definition for 'TagSubscribeRetryMaxIntervalMs'.

Step 3: Implement — in SiteRuntimeOptions.cs replace the TagSubscribeRetryIntervalMs doc comment + property with:

    /// <summary>
    /// Floor 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. As of Lever 1 (2026-08-28,
    /// MxGateway subscribe-storm prevention) it is the FLOOR of an exponential backoff:
    /// each failed/lost attempt doubles the interval (±20 % jitter) up to
    /// <see cref="TagSubscribeRetryMaxIntervalMs"/>, and a Success or a connection
    /// returning to Good quality resets it here. Default: 5000ms.
    /// </summary>
    public int TagSubscribeRetryIntervalMs { get; set; } = 5000;

    /// <summary>
    /// Ceiling (ms) of the tag-subscribe retry backoff. Must be >=
    /// <see cref="TagSubscribeRetryIntervalMs"/>; equal makes the interval fixed
    /// (pre-Lever-1 behaviour). Default: 60000ms.
    /// </summary>
    public int TagSubscribeRetryMaxIntervalMs { get; set; } = 60000;

    /// <summary>
    /// How long (ms) an outstanding tag-subscribe request blocks a new one for the
    /// same connection. While a request is in flight and younger than this, a retry
    /// tick only re-arms the watchdog instead of issuing a duplicate subscribe — this
    /// is what stops a slow gateway from receiving a fresh N-tag subscribe every
    /// interval on top of the one it is still processing. Once older than this the
    /// reply is presumed lost and a new request is allowed. Size it above the
    /// gateway's per-command deadline (MxGateway default 30 s). Default: 45000ms.
    /// </summary>
    public int TagSubscribeInFlightCeilingMs { get; set; } = 45000;

In SiteRuntimeOptionsValidator.cs, directly after the existing TagSubscribeRetryIntervalMs check:

        builder.RequireThat(options.TagSubscribeRetryMaxIntervalMs >= options.TagSubscribeRetryIntervalMs,
            $"ScadaBridge:SiteRuntime:TagSubscribeRetryMaxIntervalMs must be >= TagSubscribeRetryIntervalMs " +
            $"(was {options.TagSubscribeRetryMaxIntervalMs} vs {options.TagSubscribeRetryIntervalMs}); " +
            "it is the ceiling of the tag-subscribe retry backoff.");

        builder.RequireThat(options.TagSubscribeInFlightCeilingMs > 0,
            $"ScadaBridge:SiteRuntime:TagSubscribeInFlightCeilingMs must be greater than 0 " +
            $"(was {options.TagSubscribeInFlightCeilingMs}); it bounds how long an unanswered tag-subscribe " +
            "blocks a duplicate for the same connection.");

Step 4: Run to verify they pass

Run: dotnet test tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests --filter "FullyQualifiedName~SiteRuntimeOptionsValidatorTests" 2>&1 | tail -3 Expected: Passed! (existing DefaultOptions_AreValid still green — defaults 5000/60000/45000 satisfy both rules).

Step 5: Commit

git add src/ZB.MOM.WW.ScadaBridge.SiteRuntime/SiteRuntimeOptions.cs src/ZB.MOM.WW.ScadaBridge.SiteRuntime/SiteRuntimeOptionsValidator.cs tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/SiteRuntimeOptionsValidatorTests.cs
git commit -m "feat(site-runtime): TagSubscribeRetryMaxIntervalMs + TagSubscribeInFlightCeilingMs options (Lever 1, validated)"

Task 2: Pure backoff step `NextTagSubscribeInterval`

Classification: small Estimated implement time: ~3 min Parallelizable with: Task 1

Files:

Modify: src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/InstanceActor.cs (add a internal static method next to SendTagSubscribe, ~:1232)
Test: tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Actors/InstanceActorTagSubscribeBackoffTests.cs (create)

Step 1: Write the failing tests

using ZB.MOM.WW.ScadaBridge.SiteRuntime.Actors;

namespace ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests.Actors;

/// <summary>
/// Lever 1 (2026-08-28): the pure backoff step behind the Instance Actor's tag-subscribe
/// retry. Doubles with jitter, never below the floor, never above the ceiling; a ceiling
/// below the floor degrades to a fixed interval rather than shrinking.
/// </summary>
public class InstanceActorTagSubscribeBackoffTests
{
    private static readonly TimeSpan Floor = TimeSpan.FromSeconds(5);
    private static readonly TimeSpan Max = TimeSpan.FromSeconds(60);

    [Fact]
    public void Doubles_WithNoJitter()
    {
        var next = InstanceActor.NextTagSubscribeInterval(Floor, Floor, Max, jitterUnit: 0.5);
        Assert.Equal(TimeSpan.FromSeconds(10), next);
    }

    [Fact]
    public void Jitter_SpansPlusMinusTwentyPercent()
    {
        var low = InstanceActor.NextTagSubscribeInterval(Floor, Floor, Max, jitterUnit: 0.0);
        var high = InstanceActor.NextTagSubscribeInterval(Floor, Floor, Max, jitterUnit: 1.0);
        Assert.Equal(TimeSpan.FromSeconds(8), low);   // 10 s × 0.8
        Assert.Equal(TimeSpan.FromSeconds(12), high); // 10 s × 1.2
    }

    [Fact]
    public void NeverExceedsCeiling_EvenWithPositiveJitter()
    {
        var next = InstanceActor.NextTagSubscribeInterval(TimeSpan.FromSeconds(40), Floor, Max, jitterUnit: 1.0);
        Assert.Equal(Max, next);
    }

    [Fact]
    public void NeverBelowFloor_EvenWithNegativeJitter()
    {
        var next = InstanceActor.NextTagSubscribeInterval(TimeSpan.FromSeconds(1), Floor, Max, jitterUnit: 0.0);
        Assert.True(next >= Floor, $"{next} < floor");
    }

    [Fact]
    public void CeilingBelowFloor_DegradesToFixedFloor()
    {
        var next = InstanceActor.NextTagSubscribeInterval(Floor, Floor, TimeSpan.FromSeconds(1), jitterUnit: 0.5);
        Assert.Equal(Floor, next);
    }
}

Step 2: Run to verify they fail

Run: dotnet test tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests --filter "FullyQualifiedName~InstanceActorTagSubscribeBackoffTests" 2>&1 | tail -5 Expected: build error 'InstanceActor' does not contain a definition for 'NextTagSubscribeInterval'.

Step 3: Implement — add to InstanceActor.cs immediately above SendTagSubscribe:

    /// <summary>
    /// Pure backoff step for the tag-subscribe retry (Lever 1, 2026-08-28): double the
    /// interval just used, apply ±20 % jitter from <paramref name="jitterUnit"/> (a
    /// uniform sample in [0,1]; 0.5 = no jitter) so N instances on one connection do not
    /// re-storm in lockstep, then clamp to [floor, ceiling]. A ceiling below the floor
    /// degrades to a fixed floor interval rather than shrinking below it — same contract
    /// as <c>DataConnectionActor.NextTagResolutionInterval</c>.
    /// </summary>
    internal static TimeSpan NextTagSubscribeInterval(TimeSpan current, TimeSpan floor, TimeSpan max, double jitterUnit)
    {
        var ceiling = max < floor ? floor : max;
        if (current < floor)
            current = floor;

        var doubledMs = current.TotalMilliseconds * 2;
        var jittered = doubledMs * (0.8 + 0.4 * Math.Clamp(jitterUnit, 0.0, 1.0));
        var clamped = Math.Clamp(jittered, floor.TotalMilliseconds, ceiling.TotalMilliseconds);
        return TimeSpan.FromMilliseconds(clamped);
    }

Step 4: Run to verify they pass

Run: dotnet test tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests --filter "FullyQualifiedName~InstanceActorTagSubscribeBackoffTests" 2>&1 | tail -3 Expected: Passed! ... Total: 5. (If the SiteRuntime.Tests project lacks InternalsVisibleTo, check src/ZB.MOM.WW.ScadaBridge.SiteRuntime/ZB.MOM.WW.ScadaBridge.SiteRuntime.csproj / AssemblyInfo — DataConnectionActor.NextTagResolutionInterval is internal and tested the same way, so the SiteRuntime project very likely already has it; if not, add <InternalsVisibleTo Include="ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests" />.)

Step 5: Commit

git add src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/InstanceActor.cs tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Actors/InstanceActorTagSubscribeBackoffTests.cs
git commit -m "feat(site-runtime): pure NextTagSubscribeInterval backoff step (double, ±20% jitter, clamp)"

Classification: high-risk Estimated implement time: ~5 min Parallelizable with: none (depends on Tasks 1–2; Tasks 4–6 edit the same methods)

Files:

Modify: src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/InstanceActor.cs:180-188 (state), :1234-1255 (SendTagSubscribe), :314 (retry handler), :389-394 (PostStop)
Test: tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Actors/InstanceActorTests.cs (append after LostSubscribeResponse_IsRetried_AndSuccessStopsRetrying)

Step 1: Write the failing tests

    // --- 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);
        dclProbe.ExpectMsg<SubscribeTagsRequest>(TimeSpan.FromSeconds(3)); // second eventually re-sent

        // Success for the live id stops everything.
        var third = dclProbe.LastMessage as SubscribeTagsRequest;
        instance.Tell(new SubscribeTagsResponse(third!.CorrelationId, third.InstanceUniqueName,
            true, null, DateTimeOffset.UtcNow), dclProbe.Ref);
        dclProbe.ExpectNoMsg(TimeSpan.FromMilliseconds(800));
        _ = second;
    }

Step 2: Run to verify they fail

Run: dotnet test tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests --filter "FullyQualifiedName~InstanceActorTests.SlowSubscribe|FullyQualifiedName~InstanceActorTests.ReplyToSupersededRequest" 2>&1 | tail -8 Expected: SlowSubscribe_DoesNotStackDuplicates_WhileInFlight FAILS (ExpectNoMsg receives a SubscribeTagsRequest); SlowSubscribe_ReissuesAfterInFlightCeiling_AndOnlyOnce FAILS on the final ExpectNoMsg.

Step 3: Implement

(a) Replace the state block at :180-188 with:

    // 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<string, List<string>> _tagsByConnection = new();
    private readonly Dictionary<string, string> _pendingTagSubscribes = 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;
    }

(b) Replace SendTagSubscribe (:1226-1255, including its doc comment) with:

    /// <summary>
    /// Sends (or re-sends) the tag-subscribe request for one connection and arms a
    /// 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.
    /// <para>
    /// 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)
    {
        if (_dclManager == null) return;
        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}, next retry in {Interval})",
            _instanceUniqueName, tagPaths.Count, connectionName, correlationId, state.Interval);

        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;
    }

    /// <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);

(c) In HandleSubscribeTagsResponse (:1262-1285) — minimal change for this task; Task 4 finishes it — replace the Success branch's timer cancel:

        if (response.Success)
        {
            var state = GetTagSubscribeState(connectionName);
            if (state.InFlightCorrelationId == response.CorrelationId)
            {
                state.Timer?.Cancel();
                state.Timer = null;
                state.InFlightCorrelationId = null;
            }
            _logger.LogDebug(
                "Instance {Instance} tag subscription confirmed on connection {Connection}",
                _instanceUniqueName, connectionName);
        }

Note the guard: _pendingTagSubscribes.Remove at the top already drops truly-superseded ids (they were removed from the map at ceiling time in (b)), but a Success for the current id must clear InFlightCorrelationId so the next SendTagSubscribe (e.g. after a future failure) is not blocked by a phantom in-flight.

(d) PostStop (:391-394) — replace the timer loop:

        // Cancel any armed tag-subscribe watchdogs so they cannot fire after stop.
        foreach (var state in _tagSubscribeStates.Values)
            state.Timer?.Cancel();
        _tagSubscribeStates.Clear();

(e) Grep for any other _tagSubscribeRetryTimers reference and migrate it (expected: none besides the three sites above).

Step 4: Run to verify they pass — including the two pre-existing retry tests

Run: dotnet test tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests --filter "FullyQualifiedName~InstanceActorTests" 2>&1 | tail -5 Expected: the three new tests pass, but the two pre-existing retry tests now FAIL — correctly. Both rely on a lost/failed reply being re-sent at the 200 ms tick, and with the guard in place (default 45 s ceiling) that tick re-arms instead of re-sending (the Failure-reply path only re-sends after Task 4).

Therefore, as part of this task, amend both pre-existing retry tests so their intent survives the guard:

FailedSubscribeResponse_IsRetried: options → new SiteRuntimeOptions { TagSubscribeRetryIntervalMs = 200, TagSubscribeInFlightCeilingMs = 200 } (Task 4 makes the Failure path re-send even with a large ceiling; this keeps the test green in between).
LostSubscribeResponse_IsRetried_AndSuccessStopsRetrying: options → new SiteRuntimeOptions { TagSubscribeRetryIntervalMs = 200, TagSubscribeInFlightCeilingMs = 200 } and add a comment: // Lever 1: a lost reply is re-sent only once the outstanding request is presumed lost (ceiling), by design.

Re-run the filter above → Passed! for all five.

Step 5: Commit

git add src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/InstanceActor.cs tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Actors/InstanceActorTests.cs
git commit -m "feat(site-runtime): tag-subscribe in-flight guard — one outstanding request per connection until TagSubscribeInFlightCeilingMs (Lever 1)

LostSubscribeResponse_IsRetried_AndSuccessStopsRetrying now pins the ceiling to the retry interval: a lost reply is re-sent only once the outstanding request is presumed lost, by design."

Task 4: Failure reply clears in-flight, backs off, and re-sends at the grown interval

Classification: standard Estimated implement time: ~4 min Parallelizable with: none

Files:

Modify: src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/InstanceActor.cs (HandleSubscribeTagsResponse failure branch)
Test: tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Actors/InstanceActorTests.cs

Step 1: Write the failing tests

    // --- 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<SubscribeTagsRequest>();
        var gaps = new List<TimeSpan>();
        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<SubscribeTagsRequest>(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: the next failure after a success retries near the floor again.
        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<SubscribeTagsRequest>();
        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<SubscribeTagsRequest>(TimeSpan.FromSeconds(3));
        Assert.NotEqual(first.CorrelationId, retry.CorrelationId);
    }

Step 2: Run to verify they fail

Run: dotnet test tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests --filter "FullyQualifiedName~InstanceActorTests.RepeatedFailures|FullyQualifiedName~InstanceActorTests.FailureReply_Reissues" 2>&1 | tail -8 Expected: FailureReply_ReissuesEvenThoughGuardHasNotExpired FAILS (timeout — the guard blocks the re-send because the failure branch never clears InFlightCorrelationId); RepeatedFailures… FAILS the same way.

Step 3: Implement — replace the failure branch of HandleSubscribeTagsResponse:

        else
        {
            var state = GetTagSubscribeState(connectionName);
            if (state.InFlightCorrelationId == response.CorrelationId)
            {
                // A definite failure: the slot is free, back off, and let the watchdog
                // (already armed at the OLD interval) fire — 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 in {Interval}",
                _instanceUniqueName, connectionName, response.ErrorMessage, state.Interval);
            _ = _serviceProvider?.GetService<ISiteEventLogger>()?.LogEventAsync(
                "instance_lifecycle", "Warning", _instanceUniqueName,
                $"InstanceActor:{_instanceUniqueName}",
                $"Tag subscribe failed on connection '{connectionName}': {response.ErrorMessage} — will retry in {state.Interval}");
        }

And in the Success branch (Task 3c) add ResetTagSubscribeBackoff(state); inside the if block.

Step 4: Run to verify they pass

Run: dotnet test tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests --filter "FullyQualifiedName~InstanceActorTests" 2>&1 | tail -3 Expected: Passed! — all retry tests green. If RepeatedFailures… is flaky on gap bounds under load, widen the upper bounds only (the lower bounds are what prove backoff happened).

Step 5: Commit

git add src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/InstanceActor.cs tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Actors/InstanceActorTests.cs
git commit -m "feat(site-runtime): tag-subscribe failure reply frees the in-flight slot, backs off (×2 ±20%, capped) and Success resets to floor (Lever 1)"

Task 5: Superseded-reply drop keeps working with the state record

Classification: small Estimated implement time: ~2 min Parallelizable with: Task 6

Files:

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

This is a verification task: Task 3's ReplyToSupersededRequest_IsIgnored_AndLatestStillGoverns already covers the ceiling-superseded case. Add the failure-superseded case:

Step 1: Write the test

    [Fact]
    public void LateReplyForFailedAndReissuedRequest_IsIgnored()
    {
        var options = new SiteRuntimeOptions
        {
            TagSubscribeRetryIntervalMs = 200,
            TagSubscribeRetryMaxIntervalMs = 200,
            TagSubscribeInFlightCeilingMs = 60_000,
        };
        var (instance, dclProbe) = CreateInstanceWithDataSourcedAttribute(options);

        var first = dclProbe.ExpectMsg<SubscribeTagsRequest>();
        instance.Tell(new SubscribeTagsResponse(first.CorrelationId, first.InstanceUniqueName,
            false, "busy", DateTimeOffset.UtcNow), dclProbe.Ref);
        var second = dclProbe.ExpectMsg<SubscribeTagsRequest>(TimeSpan.FromSeconds(3));

        // Duplicate/late Success for the FIRST id: already removed from the pending map → ignored,
        // so the watchdog guarding the SECOND is untouched and still re-sends.
        instance.Tell(new SubscribeTagsResponse(first.CorrelationId, first.InstanceUniqueName,
            true, null, DateTimeOffset.UtcNow), dclProbe.Ref);
        var third = dclProbe.ExpectMsg<SubscribeTagsRequest>(TimeSpan.FromSeconds(3));
        Assert.NotEqual(second.CorrelationId, third.CorrelationId);
    }

Step 2: Run — dotnet test tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests --filter "FullyQualifiedName~LateReplyForFailedAndReissuedRequest" 2>&1 | tail -3 Expected: PASS on first run (it is a lock-in). If it fails, the bug is in Task 4's failure branch — the first id must remain removed from _pendingTagSubscribes (it is removed at the top of the handler) — fix there, don't weaken the test.

Step 3: Commit

git add tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Actors/InstanceActorTests.cs
git commit -m "test(site-runtime): lock in superseded tag-subscribe reply drop under the Lever 1 state record"

Task 6: Connection back to Good resets the backoff

Classification: standard Estimated implement time: ~4 min Parallelizable with: Task 5

Files:

Modify: src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/InstanceActor.cs:1151-1153 (HandleConnectionQualityChanged)
Test: tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Actors/InstanceActorTests.cs

Rationale: after a gateway outage the interval may sit at 60 s. The DCL re-subscribes transparently on reconnect and the Instance Actor learns of the restore only via ConnectionQualityChanged(Good). If our own subscribe is still unconfirmed at that moment, we should retry promptly, not in up to 60 s (design doc §7 "reconnect + backoff interaction").

Step 1: Write the failing test

    // --- Lever 1 (2026-08-28): Good quality resets the backoff ---

    [Fact]
    public void ConnectionBackToGood_ResetsBackoff_AndRetriesPromptly()
    {
        var options = new SiteRuntimeOptions
        {
            TagSubscribeRetryIntervalMs = 200,
            TagSubscribeRetryMaxIntervalMs = 5000,
            TagSubscribeInFlightCeilingMs = 60_000,
        };
        var (instance, dclProbe) = CreateInstanceWithDataSourcedAttribute(options);

        // Drive the interval up: 3 failures → 400, 800, 1600 (nominal).
        var req = dclProbe.ExpectMsg<SubscribeTagsRequest>();
        for (var i = 0; i < 3; i++)
        {
            instance.Tell(new SubscribeTagsResponse(req.CorrelationId, req.InstanceUniqueName,
                false, "busy", DateTimeOffset.UtcNow), dclProbe.Ref);
            req = dclProbe.ExpectMsg<SubscribeTagsRequest>(TimeSpan.FromSeconds(5));
        }
        // One more failure would nominally wait ~3.2 s. Instead the connection recovers:
        instance.Tell(new SubscribeTagsResponse(req.CorrelationId, req.InstanceUniqueName,
            false, "busy", DateTimeOffset.UtcNow), dclProbe.Ref);
        instance.Tell(new ConnectionQualityChanged("conn-1", QualityCode.Good, DateTimeOffset.UtcNow));

        var sw = System.Diagnostics.Stopwatch.StartNew();
        dclProbe.ExpectMsg<SubscribeTagsRequest>(TimeSpan.FromSeconds(2));
        Assert.True(sw.Elapsed < TimeSpan.FromMilliseconds(1000), $"retry took {sw.Elapsed} — backoff not reset");
    }

(QualityCode lives in ZB.MOM.WW.ScadaBridge.Commons.Types — check the existing usings at the top of the test file; ConnectionQualityChanged is already imported via Commons.Messages.DataConnection. If QualityCode isn't resolved, add the using for its namespace — find it with grep -rn "enum QualityCode" src/ZB.MOM.WW.ScadaBridge.Commons.)

Step 2: Run to verify it fails

Run: dotnet test tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests --filter "FullyQualifiedName~ConnectionBackToGood_ResetsBackoff" 2>&1 | tail -5 Expected: FAIL — retry arrives after ~3.2 s (or times out at 2 s).

Step 3: Implement — at the top of HandleConnectionQualityChanged, after the existing log line:

        // Lever 1 (2026-08-28): a connection returning to Good is the Instance Actor's
        // only restore signal (the DCL re-subscribes transparently). If our own subscribe
        // on it is still unconfirmed, retry at the floor now rather than in up to
        // TagSubscribeRetryMaxIntervalMs — recovery must be prompt.
        if (qualityChanged.Quality == QualityCode.Good
            && _tagSubscribeStates.TryGetValue(qualityChanged.ConnectionName, out var subState)
            && subState.Timer != null)
        {
            ResetTagSubscribeBackoff(subState);
            ArmTagSubscribeWatchdog(qualityChanged.ConnectionName, subState);
        }

(subState.Timer != null ⇔ a subscribe is unconfirmed: the timer is only ever null after a Success. If a request is in flight and young, the re-armed tick hits the guard and re-arms again at the floor — harmless.)

Step 4: Run to verify it passes

Run: dotnet test tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests --filter "FullyQualifiedName~InstanceActorTests" 2>&1 | tail -3 Expected: Passed!.

Step 5: Commit

git add src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/InstanceActor.cs tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Actors/InstanceActorTests.cs
git commit -m "feat(site-runtime): ConnectionQualityChanged(Good) resets tag-subscribe backoff to the floor (Lever 1)"

Task 7: Spec + CLAUDE.md + design-doc status

Classification: trivial Estimated implement time: ~4 min Parallelizable with: none (after Task 6)

Files:

Modify: docs/requirements/Component-SiteRuntime.md:195-201 (Instance Actor → Initialization)
Modify: docs/requirements/Component-Host.md (options table row for ScadaBridge:SiteRuntime, if the row enumerates keys — grep -n "SiteRuntimeOptions" docs/requirements/Component-Host.md)
Modify: CLAUDE.md (Architecture & Runtime bullets)
Modify: docs/plans/2026-08-28-mxgateway-subscribe-storm-prevention-design.md (§3 status line, §9 rollout)

Step 1: Spec — after item 3 of Initialization in Component-SiteRuntime.md, add:

   > **Tag-subscribe retry (S4/UA6, hardened by Lever 1 of the MxGateway subscribe-storm design, 2026-08-28).** The Instance Actor groups its data-sourced tag paths by connection and sends one `SubscribeTagsRequest` per connection, arming a per-connection watchdog *on send* so both a failed reply and a lost reply are retried. **At most one request per connection is outstanding:** a watchdog tick while the in-flight request is younger than `TagSubscribeInFlightCeilingMs` (45 s, sized above the gateway's 30 s command deadline) only re-arms — it never issues a duplicate — so a gateway that is slow to answer is not handed a fresh N-tag subscribe every interval on top of the one it is still processing (the amplifier behind the 2026-08-26 QA freeze). Past the ceiling the reply is presumed lost and a new request goes out; a late reply for the superseded id is ignored. The retry interval is exponential: floor `TagSubscribeRetryIntervalMs` (5 s), ×2 with ±20 % jitter per failed/lost attempt, capped at `TagSubscribeRetryMaxIntervalMs` (60 s); a `Success` reply or the connection returning to `Good` quality resets it to the floor. All three keys are startup-validated (`SiteRuntimeOptionsValidator`). Design: `docs/plans/2026-08-28-mxgateway-subscribe-storm-prevention-design.md` §3.

Step 2: CLAUDE.md — in Architecture & Runtime, after the "Staggered Instance Actor startup…" bullet, add:

- **Instance Actor tag-subscribe retry is guarded + backed off (Lever 1, 2026-08-28).** One outstanding `SubscribeTagsRequest` per connection until `TagSubscribeInFlightCeilingMs` (45 s); the interval doubles ±20 % jitter from `TagSubscribeRetryIntervalMs` (5 s) to `TagSubscribeRetryMaxIntervalMs` (60 s), resetting on Success or `ConnectionQualityChanged(Good)`. Pre-Lever-1 the actor re-sent a full N-tag subscribe every fixed 5 s with no in-flight check — under a slow MxGateway that stacked ~6 concurrent 244-tag subscribes per 30 s and wedged the session (QA 2026-08-26). Levers 2 (bulk supervisory advise, cross-repo) and 3 (session `command_timeout`) are still design-only: `docs/plans/2026-08-28-mxgateway-subscribe-storm-prevention-design.md`.

Step 3: Design doc — change **Status:** to Design — **Lever 1 implemented** (plan docs/plans/2026-08-28-tag-subscribe-inflight-guard-backoff.md, <commit sha>); Levers 2–3 not yet implemented. and in §9 mark step 1 done.

Step 4: Build + full SiteRuntime suite

Run: dotnet build ZB.MOM.WW.ScadaBridge.slnx 2>&1 | grep -E " error |Warning\(s\)|Build succeeded" → Build succeeded. 0 Warning(s) Run: dotnet test tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests 2>&1 | tail -3 → Passed!

Step 5: Commit

git add docs/requirements/Component-SiteRuntime.md docs/requirements/Component-Host.md CLAUDE.md docs/plans/2026-08-28-mxgateway-subscribe-storm-prevention-design.md
git commit -m "docs: tag-subscribe in-flight guard + backoff (Lever 1) — spec, CLAUDE.md, design-doc status"

Task 8: Rig redeploy + live gate

Classification: standard Estimated implement time: ~5 min (plus ~4 min deploy wall time) Parallelizable with: none

Files: none edited (verification only). Use the scadabridge-cluster-ops skill for CLI/URL.

Step 1: bash docker/deploy.sh (background), wait for /health/ready on :9001 and :9002.

Step 2: Steady-state check — pick any deployed instance on site-a and grep its subscribe log lines on the active site node:

docker logs scadabridge-site-a-a 2>&1 | grep -E "subscribing to .* tags on connection|tag subscription confirmed" | tail -6

Expected: exactly one subscribing line per (instance, connection) followed by one confirmed; no still in flight / presuming lost lines in steady state.

Step 3: Storm check (the incident shape) — stop the OPC UA infra container so subscribes fail, watch the retry cadence grow, then restore:

docker stop <opcua container from infra/docker-compose.yml>
sleep 90
docker logs --since 2m scadabridge-site-a-a 2>&1 | grep -E "tag subscribe FAILED|will retry in|presuming lost" | tail -10
docker start <opcua container>
sleep 20
docker logs --since 30s scadabridge-site-a-a 2>&1 | grep -E "quality changed to Good|tag subscription confirmed" | tail -6

Expected: will retry in values grow 00:00:05 → ~00:00:10 → ~00:00:20 → … (jittered); at most one subscribing per connection between failures (no stacking); after restore, confirmed within ~5 s of quality changed to Good (backoff reset), not tens of seconds.

Step 4: Record the result as a short postscript in the design doc §3 ("Live gate PASS : …") and commit:

git add docs/plans/2026-08-28-mxgateway-subscribe-storm-prevention-design.md
git commit -m "docs(plans): Lever 1 live gate PASS on the docker rig"

If any step fails, do NOT weaken the gate — the log lines above are the observable contract; fix the actor and re-run.

Notes for the executor

Tasks 3, 4, 6 all edit InstanceActor.cs — run them serially in one worktree; only Tasks 1/2 and 5/6 are safe in parallel (per the memory note isolation-worktree-agents-not-isolated, pre-create a worktree and cd into it; never use git stash).
Timer-based actor tests: keep bounds generous on the upper side; the lower bounds are the assertions that prove backoff/guarding.
Do not touch DataConnectionActor — its tag-resolution backoff is a different mechanism (device-side unresolved tags), deliberately left as-is.

45 KiB Raw Blame History Unescape Escape

Tag-Subscribe In-Flight Guard + Backoff (Lever 1) Implementation Plan

Background the implementer needs

Task 1: Options + validator (new keys)

Task 2: Pure backoff step NextTagSubscribeInterval

Task 3: In-flight guard — a slow subscribe issues exactly one request until the ceiling

Task 4: Failure reply clears in-flight, backs off, and re-sends at the grown interval

Task 5: Superseded-reply drop keeps working with the state record

Task 6: Connection back to Good resets the backoff

Task 7: Spec + CLAUDE.md + design-doc status

Task 8: Rig redeploy + live gate

Notes for the executor

45 KiB

Raw Blame History

Task 2: Pure backoff step `NextTagSubscribeInterval`