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
This commit is contained in:
Joseph Doherty
@@ -0,0 +1,851 @@
|
||||
# Tag-Subscribe In-Flight Guard + Backoff (Lever 1) Implementation Plan
|
||||
|
||||
> **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`:
|
||||
|
||||
```csharp
|
||||
// --- 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:
|
||||
|
||||
```csharp
|
||||
/// <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:
|
||||
|
||||
```csharp
|
||||
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**
|
||||
|
||||
```bash
|
||||
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**
|
||||
|
||||
```csharp
|
||||
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`:
|
||||
|
||||
```csharp
|
||||
/// <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**
|
||||
|
||||
```bash
|
||||
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)"
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
### Task 3: In-flight guard — a slow subscribe issues exactly one request until the ceiling
|
||||
|
||||
**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**
|
||||
|
||||
```csharp
|
||||
// --- 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:
|
||||
|
||||
```csharp
|
||||
// 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:
|
||||
|
||||
```csharp
|
||||
/// <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:
|
||||
|
||||
```csharp
|
||||
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:
|
||||
|
||||
```csharp
|
||||
// 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**
|
||||
|
||||
```bash
|
||||
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**
|
||||
|
||||
```csharp
|
||||
// --- 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`:
|
||||
|
||||
```csharp
|
||||
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**
|
||||
|
||||
```bash
|
||||
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**
|
||||
|
||||
```csharp
|
||||
[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**
|
||||
|
||||
```bash
|
||||
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**
|
||||
|
||||
```csharp
|
||||
// --- 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 `using`s 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:
|
||||
|
||||
```csharp
|
||||
// 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**
|
||||
|
||||
```bash
|
||||
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:
|
||||
|
||||
```markdown
|
||||
> **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:
|
||||
|
||||
```markdown
|
||||
- **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**
|
||||
|
||||
```bash
|
||||
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:
|
||||
|
||||
```bash
|
||||
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:
|
||||
|
||||
```bash
|
||||
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 <date>: …") and commit:
|
||||
|
||||
```bash
|
||||
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.
|
||||
@@ -0,0 +1,14 @@
|
||||
{
|
||||
"planPath": "docs/plans/2026-08-28-tag-subscribe-inflight-guard-backoff.md",
|
||||
"tasks": [
|
||||
{"id": 1, "subject": "Task 1: Options + validator (new keys)", "status": "pending"},
|
||||
{"id": 2, "subject": "Task 2: Pure backoff step NextTagSubscribeInterval", "status": "pending"},
|
||||
{"id": 3, "subject": "Task 3: In-flight guard", "status": "pending", "blockedBy": [1, 2]},
|
||||
{"id": 4, "subject": "Task 4: Failure reply frees slot, backs off, Success resets", "status": "pending", "blockedBy": [3]},
|
||||
{"id": 5, "subject": "Task 5: Superseded-reply lock-in test", "status": "pending", "blockedBy": [4]},
|
||||
{"id": 6, "subject": "Task 6: ConnectionQualityChanged(Good) resets backoff", "status": "pending", "blockedBy": [4]},
|
||||
{"id": 7, "subject": "Task 7: Spec + CLAUDE.md + design-doc status", "status": "pending", "blockedBy": [5, 6]},
|
||||
{"id": 8, "subject": "Task 8: Rig redeploy + live gate", "status": "pending", "blockedBy": [7]}
|
||||
],
|
||||
"lastUpdated": "2026-08-28T18:00:00Z"
|
||||
}
|
||||
Reference in New Issue
Block a user