Compare commits
11 Commits
a9b23556eb
...
afb5180ce0
| Author | SHA1 | Date | |
|---|---|---|---|
 Joseph Doherty
|
afb5180ce0 | ||
|
Joseph Doherty
|
c52f14afca | ||
|
Joseph Doherty
|
0589486f4e | ||
|
Joseph Doherty
|
150c3f60d8 | ||
|
Joseph Doherty
|
ab08ecd3d3 | ||
|
Joseph Doherty
|
b6b401c879 | ||
|
Joseph Doherty
|
029d39fca2 | ||
|
Joseph Doherty
|
2f471dd7a1 | ||
|
Joseph Doherty
|
6f7c6ad877 | ||
|
Joseph Doherty
|
e39cca093b | ||
|
Joseph Doherty
|
dce52ad087 |
@@ -70,6 +70,7 @@ spec for each is `docs/requirements/Component-<Name>.md`, and `README.md` carrie
|
||||
- Site-wide Akka stream for attribute value and alarm state changes with per-subscriber buffering.
|
||||
- Instance Actors serialize all state mutations (Akka actor model); concurrent scripts produce interleaved side effects.
|
||||
- Staggered Instance Actor startup on failover to prevent reconnection storms.
|
||||
- **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). Consequence to know: a *lost* reply is now re-sent after the ceiling (45 s), not the 5 s interval — a *failed* reply still retries at the backed-off interval. 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`.
|
||||
- Explicit supervision strategies: Resume for coordinator actors, Stop for short-lived execution actors.
|
||||
|
||||
### Data & Communication
|
||||
|
||||
@@ -0,0 +1,179 @@
|
||||
# MxGateway subscribe-storm prevention — design
|
||||
|
||||
**Date:** 2026-08-28
|
||||
**Status:** Design — **Lever 1 implemented** (plan `docs/plans/2026-08-28-tag-subscribe-inflight-guard-backoff.md`, branch commits 6f7c6ad8..ab08ecd3; Levers 2–3 not yet implemented) — **reviewed against both repos 2026-08-28** (ScadaBridge `main` @ `a9b23556`, MxAccessGateway @ `74ba889`): no part of this design has landed in either repo; corrections from that review are folded in below (Lever 3 was re-scoped — see §5).
|
||||
**Scope:** ScadaBridge (`SiteRuntime`, `DataConnectionLayer`) **and** MxAccessGateway (`~/Desktop/MxAccessGateway`: `Contracts` proto, `Worker`, `Server`)
|
||||
**Motivating incident:** 2026-08-26 QA (vq03/vq04) — Z28061's 244 attributes went Bad/frozen for ~20 min. Root cause was a **transient MxAccessGateway subscribe wedge under deploy/subscribe churn**, not ScadaBridge or Galaxy. No dedicated incident record exists for 2026-08-26 yet — the analysis lives in this doc. Related prior record: `docs/deployment/deployment-records/2026-08-20-wonder-app-vq03-vq04.md` §4, the earlier **ScadaBridge DCL** hotfix `da0bfe2b` on the same connection (MxGateway adapter treating a graceful event-stream end as a disconnect — a ScadaBridge commit, not a gateway one; the 2026-08-26 freeze is a *different* failure mode, see §1).
|
||||
|
||||
---
|
||||
|
||||
## 1. Problem statement
|
||||
|
||||
Two independent defects compound into a self-sustaining freeze:
|
||||
|
||||
1. **Amplifier (ScadaBridge).** `InstanceActor.SendTagSubscribe` arms a **fixed 5 s** retry timer (`SiteRuntimeOptions.TagSubscribeRetryIntervalMs = 5000`) **on send**, with **no in-flight guard and no backoff**. When a `SubscribeBulk` round-trip takes longer than 5 s — which it does the moment the gateway is under load — a *new* full 244-tag subscribe is issued every 5 s while the previous one is still outstanding. With two instances (Z28061 + Z28062) this stacks ~6 concurrent 244-tag subscribes per 30 s window.
|
||||
- `src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/InstanceActor.cs:1234` (`SendTagSubscribe`, called from `SubscribeToDcl` :1224 and the retry handler), `:1250-1254` (timer arm — the only "guard" is that re-arming overwrites the previous per-connection timer; nothing checks `_pendingTagSubscribes` before issuing), `:1264` (stale-correlation drop), `:1269` (cancel-on-success), `:314` (`RetryTagSubscribe` handler), `:391-394` (`PostStop` cancels timers). State fields `_tagsByConnection`/`_pendingTagSubscribes`/`_tagSubscribeRetryTimers` at `:185-188`. `SiteRuntimeOptions.cs:120` (`TagSubscribeRetryIntervalMs = 5000`, doc comment explicitly says "fixed interval"); `SiteRuntimeOptionsValidator.cs:77-80` checks only `> 0`.
|
||||
|
||||
2. **Load (cross-repo).** In supervisory mode (`WriteUserId == 0`, how QA runs), one ScadaBridge subscribe = **1 `AddItemBulk` RPC + ~244 per-item `AdviseSupervisory` RPCs** (pipelined ×16). Each `AdviseSupervisory` is a separate STA command on the worker's single-threaded apartment. The gateway's per-command deadline is **30 s** (`MxGateway:Sessions:DefaultCommandTimeoutSeconds`), so under contention these bulk/per-item commands hit the deadline, cascade `DeadlineExceeded`, and no advise is ever established → no `OnDataChange` reaches ScadaBridge → all tags Bad.
|
||||
- ScadaBridge: `src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Adapters/RealMxGatewayClient.cs:108-160` (`SubscribeBulkAsync`, supervisory branch).
|
||||
- Gateway (canonical proto `src/ZB.MOM.WW.MxGateway.Contracts/Protos/mxaccess_gateway.proto`; the copies under `clients/` and `build/` are generated mirrors): `SubscribeBulkCommand` (:325-328, `server_handle=1`, `tag_addresses=2`, no field 3) does AddItem + **plain** `Advise` per tag in one STA pass — `Worker/MxAccess/MxAccessSession.cs:628-652` (`SubscribeBulk`, with `AppendRemoveItemCleanup` rollback), `Worker/MxAccess/MxAccessCommandExecutor.cs:646` (`ExecuteSubscribeBulk`). The only supervisory entry point is the per-item `AdviseSupervisoryCommand` (proto :220-223, oneof 17; `MxAccessSession.AdviseSupervisory` :319-331). Note an `AdviseItemBulkCommand` (proto :312, oneof 29, `repeated int32 item_handles`) already exists for the *plain* kind — there is simply no supervisory counterpart, which is why ScadaBridge fans out per item. The STA really is one command at a time: `Worker/Sta/StaRuntime.cs:58-63` (single STA thread), `StaCommandDispatcher.DrainAsync` :233-268 (awaits each command fully before dequeuing the next).
|
||||
|
||||
**Proof it is not MXAccess/Galaxy:** a fresh `LMXProxy` 32-bit STA probe did AddItem + AdviseSupervisory for every sampled tag in 0–5 ms while both worker processes sat < 5 % CPU. The bottleneck is command-path fan-out and retry stacking, not the OT layer.
|
||||
|
||||
### Goal / non-goals
|
||||
|
||||
- **Goal:** a subscribe or reconnect storm on one instance can no longer wedge the gateway session or freeze data; recovery is prompt and self-correcting.
|
||||
- **Non-goal:** changing the alarm/event streaming path (`SiteStreamService`, `da0bfe2b`), the write path, or anything about Galaxy/MXAccess itself.
|
||||
- **Non-goal:** all-or-nothing subscribe semantics; per-tag partial success is preserved.
|
||||
|
||||
---
|
||||
|
||||
## 2. Approach (chosen: full fix)
|
||||
|
||||
Three levers were considered; the full fix combines the two that address root cause, and folds in the third as a bounded safety knob:
|
||||
|
||||
| Lever | What | Repo | Role |
|
||||
|---|---|---|---|
|
||||
| **1 — retry backoff + in-flight guard** | Stop re-sending a subscribe while one is in flight; exponential backoff with jitter on genuine failure | ScadaBridge | Removes the **amplifier** |
|
||||
| **2 — bulk supervisory-advise RPC** | `SubscribeBulkCommand.supervisory` flag → worker does AddItem + `AdviseSupervisory` in one STA pass; collapses ~244 RPCs → 1 | Cross-repo, **capability-negotiated** | Removes the **load** |
|
||||
| **3 — session command timeout** | ScadaBridge sets an explicit `command_timeout` at `OpenSession` (the gateway's only existing override — it is **per-session**, not per-command; see §5) | ScadaBridge | Bounded safety margin |
|
||||
|
||||
Levers 1 and 2 are each sufficient to break the feedback loop; together they are defense-in-depth (1 caps concurrency, 2 cuts the per-subscribe cost by ~244×). Lever 3 is a cheap knob that stops treating a merely-slow (but healthy) bulk op as a failure — with the caveat, found in review, that the gateway has no per-command override, so the knob is session-wide.
|
||||
|
||||
**Rejected as standalone:** raising `DefaultCommandTimeoutSeconds` globally (band-aid — lets the queue back up *longer*); an unnegotiated proto bump for Lever 2 (an old gateway would silently do a **plain** advise, breaking writes on a `WriteUserId==0` connection).
|
||||
|
||||
---
|
||||
|
||||
## 3. Lever 1 — ScadaBridge retry backoff + in-flight guard
|
||||
|
||||
**Where:** `InstanceActor` (`SiteRuntime`). Per-connection subscribe state already lives here (`_tagSubscribeRetryTimers`, `_pendingTagSubscribes`).
|
||||
|
||||
**Change:**
|
||||
- Track an **in-flight** flag per connection. `SendTagSubscribe` becomes: if a subscribe for this connection is already outstanding **and** not past a hard "lost reply" ceiling, do **not** issue a new one — just re-arm the watchdog. This alone stops the 5 s stacking.
|
||||
- Replace the fixed 5 s re-arm with **exponential backoff + jitter**: floor `TagSubscribeRetryIntervalMs` (keep 5 s default), doubling to a new `TagSubscribeRetryMaxIntervalMs` (default 60 s), reset to the floor on a `Success` reply. Mirrors the existing DCL pattern for unresolved tags (`DataConnectionOptions.cs:17,24` `TagResolutionRetryInterval` 10 s / `TagResolutionRetryMaxInterval` 5 min, validated floor `> 0` and max `>= floor` at `DataConnectionOptionsValidator.cs:21-22,39-41`; machinery in `DataConnectionActor.cs` — `ScheduleTagResolutionRetry` :2393-2405 with its `IsTimerActive` guard, `ResetTagResolutionBackoff` :2409, `BackOffTagResolution` :2417-2419, pure step `NextTagResolutionInterval` ~:2422) — reuse the shape for consistency.
|
||||
- **Lost-reply safety:** because a reply can be genuinely lost, the in-flight guard must expire. Keep a `TagSubscribeInFlightCeilingMs` (default = current gateway command timeout + margin, e.g. 45 s) after which a new subscribe is allowed even if the old correlation never answered. This preserves today's "retry fires whether the response fails or is lost" guarantee without the 5 s stacking.
|
||||
|
||||
**New options** (`SiteRuntimeOptions` + validator):
|
||||
- `TagSubscribeRetryMaxIntervalMs` (default 60000; validate `>= TagSubscribeRetryIntervalMs`).
|
||||
- `TagSubscribeInFlightCeilingMs` (default 45000; validate `> 0`).
|
||||
|
||||
**Invariants preserved:** a superseded correlation id is still ignored (`HandleSubscribeTagsResponse` :1262-1264 stale-reply drop); a `Success` still cancels the timer (:1269) and resets backoff; `PostStop` still cancels all timers (`:391-394`).
|
||||
|
||||
|
||||
> **Live gate PASS 2026-08-28 (docker rig, main @ 0589486f).** OPC UA sim stopped and the instance's site node restarted so the boot-time subscribe could not be answered (failover landed `rig-probe-1` on site-a-b): the log shows exactly **one** outstanding `SubscribeTagsRequest` per in-flight ceiling — `unanswered for 45091ms … presuming lost, re-sending`, then `47130ms` — with the backoff climbing `next retry in 00:00:05 → 00:00:09.415 → 00:00:20.369` (jittered), i.e. 3 requests in ~110 s where the pre-Lever-1 actor would have issued ~22. Sim restarted → DCL `Entering Connected state` in 4 s → all three attributes `Good` with fresh timestamps within ~95 s (`debug snapshot --id 1492`). Steady state on a healthy node: one `subscribing` line per (instance, connection), no `still in flight` / `presuming lost` at all. Note for future gates: an OPC UA outage *alone* never reaches this path — the DCL reconnects and re-subscribes transparently and the Instance Actor's subscribe stays confirmed; you must make the subscribe itself go unanswered (node restart / redeploy while the source is down).
|
||||
|
||||
---
|
||||
|
||||
## 4. Lever 2 — bulk supervisory-advise RPC (capability-negotiated)
|
||||
|
||||
**Contract (`mxaccess_gateway.proto`), additive only:**
|
||||
```proto
|
||||
message SubscribeBulkCommand {
|
||||
int32 server_handle = 1;
|
||||
repeated string tag_addresses = 2;
|
||||
bool supervisory = 3; // NEW — advise supervisory instead of plain advise
|
||||
}
|
||||
```
|
||||
Field 3 is unused today (verified); additive, field numbers never reused (house rule). Regen is the manual vendored-proto toggle both repos already use. **Alternative considered:** a new `AdviseSupervisoryItemBulkCommand` mirroring the existing plain `AdviseItemBulkCommand` (oneof 29) — rejected because it keeps the two-RPC AddItemBulk+AdviseBulk shape and two STA commands; the flag gives one RPC / one STA pass and reuses `SubscribeBulkReply` unchanged.
|
||||
|
||||
**Capability negotiation (the back-compat guarantee):** the gateway already advertises a capability list on `OpenSessionReply.capabilities` (field 5) — today exactly nine strings: `unary-open-session`, `unary-close-session`, `unary-invoke`, `server-stream-events`, `bulk-subscribe-commands`, `bulk-read-commands`, `bulk-write-commands`, `unary-acknowledge-alarm`, `server-stream-active-alarms` (`Server/Grpc/MxAccessGatewayService.cs:59-67`; `bulk-subscribe-supervisory` does not exist anywhere yet). The reply also carries `gateway_protocol_version` (field 8) — a capability string is still the right gate (feature-keyed, not version-keyed). Add:
|
||||
```
|
||||
reply.Capabilities.Add("bulk-subscribe-supervisory");
|
||||
```
|
||||
ScadaBridge reads the capability set at session open and only sets `supervisory = true` when `bulk-subscribe-supervisory` is present. Against an older gateway the capability is absent → ScadaBridge keeps today's `AddItemBulk` + per-item `AdviseSupervisory` path. **This is the correctness gate:** a plain advise on a `WriteUserId==0` connection would silently break writes, so we must never send `supervisory=true` blind.
|
||||
- ScadaBridge does not currently retain the capability list in the DCL adapter — verified: `RealMxGatewayClient.ConnectAsync` (:83-84) does `OpenSessionAsync` then `RegisterAsync` and no `Capabilities` reference exists anywhere in `DataConnectionLayer`. This design adds: capture `OpenSessionReply.capabilities` into the `RealMxGatewayClient` at open, expose a `SupportsSupervisorySubscribeBulk` predicate.
|
||||
|
||||
**Worker (`Worker/MxAccess/…`):**
|
||||
- `MxAccessSession.SubscribeBulk` gains a `supervisory` parameter; when true it calls the MXAccess `AdviseSupervisory` (and registers `MxAccessAdviceKind.Supervisory`, exactly as the per-item path at :319-331 does) instead of plain `Advise` at :651-652 inside the same AddItem+advise loop. Per-item try/catch and `AppendRemoveItemCleanup` rollback are unchanged (partial success preserved). Long bulk commands hold the STA — keep the existing `StaRuntime.PumpPendingMessages()` (:94) behaviour the plain bulk path relies on.
|
||||
- `ExecuteSubscribeBulk` passes `subscribeBulkCommand.Supervisory` through (`MxAccessCommandExecutor.cs:646`). Reply shape (`SubscribeBulkReply`, per-item handles) is unchanged.
|
||||
|
||||
**ScadaBridge (`RealMxGatewayClient.SubscribeBulkAsync`, `:108`):**
|
||||
- Supervisory branch (`UseSupervisoryAdvise` :53, `_writeUserId == 0`): if the gateway advertises the capability, issue **one** `SubscribeBulkAsync(_serverHandle, tagPaths, supervisory: true, ct)` instead of `AddItemBulkAsync` (:128) + the `BulkPipeline.RunAsync` (:136-146) of per-item `EnsureSupervisoryAdvisedAsync` (:142, method :591; parallelism `MxSupervisoryAdviseParallelism` = 16, `DataConnectionOptions.cs:73`). Result mapping (`MapSubscribeResults`, handle bookkeeping, failed-item handle eviction) reuses the existing plain-mode path.
|
||||
- Fallback branch unchanged. Delete the now-superseded "Cross-repo follow-up" comment (`RealMxGatewayClient.cs:126-127`) since this design fulfills it.
|
||||
- Plain-advise mode (a configured write user) already used single-RPC `SubscribeBulkAsync` and is untouched.
|
||||
|
||||
**Net effect:** supervisory subscribe drops from `1 + N` RPCs and `N` STA commands to **1 RPC / 1 STA command** for N tags.
|
||||
|
||||
---
|
||||
|
||||
## 5. Lever 3 — explicit session command timeout (re-scoped in review)
|
||||
|
||||
**Correction from the 2026-08-28 review:** the gateway has **no per-command timeout override**. `MxCommandRequest` (proto :89-93) carries only `session_id`, `client_correlation_id`, `command`, and `MxAccessGrpcRequestValidator.ValidateInvoke` (:38-54) has no timeout handling. The only override is **per session**: `OpenSessionRequest.command_timeout` (field 4, `google.protobuf.Duration`), validated `> 0` (`MxAccessGrpcRequestValidator.cs:15`), resolved by `SessionManager.ResolveCommandTimeout` (:583, called from :528) to `MxGateway:Sessions:DefaultCommandTimeoutSeconds` (default 30, `SessionOptions.cs:8`) when absent, then captured on `GatewaySession.CommandTimeout` (:223,293) and applied to **every** invoke on that session (:1151). The reply echoes the resolved value as `OpenSessionReply.default_command_timeout` (field 6). ScadaBridge's DCL never sets it and never reads it back (verified — no `CommandTimeout` anywhere in `DataConnectionLayer`; only `_readTimeoutMs` for reads).
|
||||
|
||||
Two ways to get a bulk-sized margin; **(a) is chosen**:
|
||||
|
||||
- **(a) Session-scoped, ScadaBridge-only (chosen).** New `DataConnectionOptions.MxCommandTimeout` (default 60 s, validated `> 0`), sent as `OpenSessionRequest.command_timeout` in `RealMxGatewayClient.ConnectAsync`. Trade: it governs **every** command on the session, writes included — a wedged write would take up to 60 s instead of 30 s to fault. Acceptable: ScadaBridge's own `WriteTimeout` still bounds the script-facing write, and the gateway deadline is the backstop, not the primary. Log the echoed `default_command_timeout` at open so a gateway that clamps or ignores the request is visible.
|
||||
- **(b) Per-invoke override (cross-repo, deferred).** Additive `google.protobuf.Duration command_timeout = 4` on `MxCommandRequest`, honoured in `GatewaySession` invoke resolution, capability-advertised (`per-command-timeout`). Cleaner but a second wire change for a margin that Lever 2 makes mostly moot (a single-pass bulk subscribe completes well under a second). Revisit only if (a)'s write-side widening proves a problem.
|
||||
|
||||
This remains a margin, not a fix — it only stops a merely-slow-but-alive gateway from being treated as failed and re-storming.
|
||||
|
||||
---
|
||||
|
||||
## 6. Data flow (after)
|
||||
|
||||
```
|
||||
InstanceActor.SendTagSubscribe(conn)
|
||||
└─ in-flight guard: skip if a subscribe for conn is outstanding & within ceiling
|
||||
└─ else: 1 SubscribeTagsRequest ─▶ DataConnectionActor
|
||||
└─ RealMxGatewayClient.SubscribeBulkAsync(tagPaths)
|
||||
├─ gateway advertises bulk-subscribe-supervisory?
|
||||
│ yes → 1 RPC SubscribeBulk(supervisory=true) [session CommandTimeout = MxCommandTimeout, set at OpenSession]
|
||||
│ └─ worker: AddItem + AdviseSupervisory per tag, ONE STA pass
|
||||
│ no → legacy AddItemBulk + N×AdviseSupervisory (fallback)
|
||||
└─ per-tag results ─▶ DCL resolves handles ─▶ OnDataChange flows
|
||||
└─ Success reply → cancel timer, reset backoff to floor
|
||||
└─ Failure/lost → backoff × jitter, re-arm (bounded by max interval)
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## 7. Error handling & edge cases
|
||||
|
||||
- **Old gateway / capability absent:** ScadaBridge falls back to today's path; no behavior change, writes stay correct.
|
||||
- **Partial subscribe failure:** unchanged — per-tag success/failure preserved end to end; failed items evict their handle mapping so a later retry re-adds cleanly.
|
||||
- **Lost subscribe reply:** in-flight ceiling releases the guard after `TagSubscribeInFlightCeilingMs`; a fresh subscribe then issues. No permanent stall.
|
||||
- **Gateway restart mid-subscribe (the incident trigger):** connection-level fault path (`DataConnectionActor` reconnect state machine) is unchanged; on reconnect the first subscribe issues immediately (backoff reset), and with the in-flight guard the reconnect storm cannot stack.
|
||||
- **Reconnect + backoff interaction:** a connection restore must reset backoff to the floor so recovery is prompt (don't inherit a stretched interval from the pre-fault failures).
|
||||
|
||||
---
|
||||
|
||||
## 8. Testing
|
||||
|
||||
**ScadaBridge (Lever 1):**
|
||||
- Unit (`InstanceActor` tests): a slow/never-answered subscribe issues exactly **one** in-flight request until the ceiling, not one per interval; backoff doubles on repeated failure and resets on success; a connection restore resets backoff.
|
||||
- Option validation: `TagSubscribeRetryMaxIntervalMs >= TagSubscribeRetryIntervalMs`; ceiling `> 0`.
|
||||
|
||||
**Cross-repo (Lever 2):**
|
||||
- Worker unit: `SubscribeBulk(supervisory:true)` calls `AdviseSupervisory` (fake MXAccess session asserts advise kind); partial-failure and cleanup unchanged.
|
||||
- Gateway: `OpenSessionReply.capabilities` includes `bulk-subscribe-supervisory`.
|
||||
- ScadaBridge adapter: with capability present → single `SubscribeBulk(supervisory=true)`; absent → legacy fan-out. Handle bookkeeping identical across both paths (round-trip test).
|
||||
- Proto: additive-only guard (field 3 added, none renumbered).
|
||||
|
||||
**Lever 3:** `RealMxGatewayClient.ConnectAsync` sends `OpenSessionRequest.command_timeout = MxCommandTimeout`; the fake gateway asserts the value and the echoed `default_command_timeout` is logged; option validated `> 0`.
|
||||
|
||||
**Live gate (QA rig, after deploy — not part of this doc's execution):** reproduce the incident shape — redeploy Z28061 several times in quick succession while watching `health site site-a`; confirm `successfullyResolved` tracks to full without a Bad/frozen window, gateway log shows no `AddItemBulk/AdviseSupervisory timed out`, and (with Lever 2 active) the per-subscribe RPC count on the gateway session drops ~244×.
|
||||
|
||||
---
|
||||
|
||||
## 9. Rollout & sequencing
|
||||
|
||||
Levers are independent and independently shippable; recommended order:
|
||||
|
||||
1. **Lever 1 (ScadaBridge-only) — DONE 2026-08-28.** Highest value / lowest blast radius — breaks the amplifier immediately; no gateway change. Ships and deploys as a ScadaBridge binary swap.
|
||||
2. **Lever 2 (gateway first, then ScadaBridge).** Deploy the gateway side (new capability + worker flag) before ScadaBridge starts requesting it — capability negotiation makes the order safe either way, but gateway-first means the optimization lights up as soon as ScadaBridge ships. Both are additive/wire-compatible.
|
||||
3. **Lever 3 (ScadaBridge-only).** Fold in with Lever 1 or Lever 2's ScadaBridge change; pure config surface.
|
||||
|
||||
No EF migrations, no schema, no new ports. MxAccessGateway proto regen uses the existing manual vendored-proto toggle in both repos.
|
||||
|
||||
**Cross-repo index:** update the ScadaBridge entry in `../scadaproj/CLAUDE.md` if the OPC-UA/MxGateway wire relationship summary changes (per repo rule); update `MxAccessGateway`'s own capability list docs for `bulk-subscribe-supervisory`.
|
||||
|
||||
---
|
||||
|
||||
## 10. Open questions
|
||||
|
||||
- **Backoff defaults:** 5 s floor → 60 s ceiling proposed; confirm the ceiling is acceptable for reconnect-recovery latency (a restore resets to floor, so steady-state recovery is unaffected).
|
||||
- **Lever 3 value:** 60 s `MxCommandTimeout` assumes Lever 2 lands; without Lever 2 the legacy fan-out is N separate 30 s-bounded commands, so the session timeout barely matters for subscribe — it is the *write-side* widening (§5 trade) that decides whether 60 s is right. Size against the fallback path if shipping Lever 3 before Lever 2.
|
||||
- **Incident record:** write `docs/deployment/deployment-records/2026-08-26-wonder-app-vq03-vq04-freeze.md` (or fold into the next QA deploy record) so the 2026-08-26 analysis is not only in a design doc.
|
||||
- **`NativeAlarmActor` is the same amplifier, still unguarded (final review of Lever 1, 2026-08-28).** `NativeAlarmActor.SendSubscribe` (`SiteRuntime/Actors/NativeAlarmActor.cs:147-165`) arms a fixed `NativeAlarmRetryIntervalMs × 2` timer on send with no in-flight check and no backoff, and `HandleSubscribeResponse` (`:319-337`) never compares the correlation id — so a stale reply can cancel the timer guarding a newer request. One actor per native alarm source per instance, over the same DCL connections that wedged on 2026-08-26. **Follow-up: port Lever 1's `TagSubscribeState` shape to it** (or a Gitea issue) before the next incident finds it.
|
||||
- **Cancel-race residue (Lever 1 M5):** `ICancelable.Cancel()` cannot retract a `RetryTagSubscribe` already in the mailbox, so a tick that races a Success issues one spurious full subscribe (self-healing, bounded to one). A `Confirmed` flag on `TagSubscribeState` would close it in ~4 lines; deferred until it is observed.
|
||||
- **Should the in-flight guard also live in `DataConnectionActor`** (coalescing duplicate concurrent subscribe requests for the same connection across instances), as a second layer? Deferred — `InstanceActor` is where the storm originates and is sufficient; revisit only if multi-instance coalescing proves necessary.
|
||||
@@ -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,66 @@
|
||||
{
|
||||
"planPath": "docs/plans/2026-08-28-tag-subscribe-inflight-guard-backoff.md",
|
||||
"tasks": [
|
||||
{
|
||||
"id": 1,
|
||||
"subject": "Task 1: Options + validator (new keys)",
|
||||
"status": "completed"
|
||||
},
|
||||
{
|
||||
"id": 2,
|
||||
"subject": "Task 2: Pure backoff step NextTagSubscribeInterval",
|
||||
"status": "completed"
|
||||
},
|
||||
{
|
||||
"id": 3,
|
||||
"subject": "Task 3: In-flight guard",
|
||||
"status": "completed",
|
||||
"blockedBy": [
|
||||
1,
|
||||
2
|
||||
]
|
||||
},
|
||||
{
|
||||
"id": 4,
|
||||
"subject": "Task 4: Failure reply frees slot, backs off, Success resets",
|
||||
"status": "completed",
|
||||
"blockedBy": [
|
||||
3
|
||||
]
|
||||
},
|
||||
{
|
||||
"id": 5,
|
||||
"subject": "Task 5: Superseded-reply lock-in test",
|
||||
"status": "completed",
|
||||
"blockedBy": [
|
||||
4
|
||||
]
|
||||
},
|
||||
{
|
||||
"id": 6,
|
||||
"subject": "Task 6: ConnectionQualityChanged(Good) resets backoff",
|
||||
"status": "completed",
|
||||
"blockedBy": [
|
||||
4
|
||||
]
|
||||
},
|
||||
{
|
||||
"id": 7,
|
||||
"subject": "Task 7: Spec + CLAUDE.md + design-doc status",
|
||||
"status": "completed",
|
||||
"blockedBy": [
|
||||
5,
|
||||
6
|
||||
]
|
||||
},
|
||||
{
|
||||
"id": 8,
|
||||
"subject": "Task 8: Rig redeploy + live gate",
|
||||
"status": "completed",
|
||||
"blockedBy": [
|
||||
7
|
||||
]
|
||||
}
|
||||
],
|
||||
"lastUpdated": "2026-08-28T19:06:22Z"
|
||||
}
|
||||
@@ -196,6 +196,9 @@ flowchart TD
|
||||
1. Load all attribute values from the flattened configuration (static defaults).
|
||||
2. Set quality to **uncertain** for all attributes that have a data source reference. Static attributes (no data source reference) have quality **good**. The uncertain quality persists until the first value update arrives from the Data Connection Layer, distinguishing "not yet received" from "known good" or "connection lost."
|
||||
3. Register data source references with the Data Connection Layer for subscriptions.
|
||||
|
||||
> **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 — but the `Good` reset does **not** release an outstanding request (it still owns the connection until the ceiling), so it shortens the wait only when nothing is in flight; and once the backed-off interval exceeds the ceiling the guard is inert on ticks and the interval itself is the rate limit (one re-send per 60 s at the defaults). All three keys are startup-validated (`SiteRuntimeOptionsValidator`). Design: `docs/plans/2026-08-28-mxgateway-subscribe-storm-prevention-design.md` §3; plan: `docs/plans/2026-08-28-tag-subscribe-inflight-guard-backoff.md`.
|
||||
|
||||
4. Create child Script Actors (one per script defined on the instance).
|
||||
5. Create child Alarm Actors (one per alarm defined on the instance).
|
||||
|
||||
|
||||
@@ -178,14 +178,32 @@ public class InstanceActor : ReceiveActor
|
||||
// fan out to all of them — not just the last one registered.
|
||||
private readonly Dictionary<string, List<string>> _tagPathToAttributes = new();
|
||||
|
||||
// Tag-subscription retry state (S4/UA6): tag paths grouped by connection name
|
||||
// (built once in SubscribeToDcl), the in-flight correlation-id → connection
|
||||
// map, and a per-connection retry timer. Arming the timer on *send* closes both
|
||||
// the failed-response and the lost-response gaps with one mechanism; the timer
|
||||
// is cancelled when a Success reply arrives.
|
||||
// Tag-subscription retry state (S4/UA6 + Lever 1 2026-08-28): tag paths grouped by
|
||||
// connection name (built once in SubscribeToDcl), the in-flight correlation-id →
|
||||
// connection map, and per-connection retry state. Arming the watchdog on *send*
|
||||
// closes both the failed-response and the lost-response gaps with one mechanism;
|
||||
// it is cancelled when a Success reply arrives. Lever 1 adds the in-flight guard
|
||||
// (a tick while a young request is outstanding re-arms instead of re-sending) and
|
||||
// exponential backoff (see NextTagSubscribeInterval) so a slow gateway is never
|
||||
// handed a fresh N-tag subscribe on top of the one it is still processing.
|
||||
// SubscribeToDcl has a single caller today (PreStart); the prune at its top is defensive, so a
|
||||
// future second call site rebuilds the retry state and both tag maps together instead of stranding them.
|
||||
private readonly Dictionary<string, List<string>> _tagsByConnection = new();
|
||||
private readonly Dictionary<string, string> _pendingTagSubscribes = new();
|
||||
private readonly Dictionary<string, ICancelable> _tagSubscribeRetryTimers = new();
|
||||
private readonly Dictionary<string, TagSubscribeState> _tagSubscribeStates = new();
|
||||
|
||||
/// <summary>Per-connection tag-subscribe retry bookkeeping (Lever 1).</summary>
|
||||
private sealed class TagSubscribeState
|
||||
{
|
||||
/// <summary>Correlation id of the outstanding request, null when none is in flight.</summary>
|
||||
public string? InFlightCorrelationId;
|
||||
/// <summary>When the outstanding request was sent (monotonic, for the ceiling check).</summary>
|
||||
public long InFlightSentTicks;
|
||||
/// <summary>Interval the NEXT retry will use; starts at the floor, grows on failure.</summary>
|
||||
public TimeSpan Interval;
|
||||
/// <summary>The armed watchdog for this connection.</summary>
|
||||
public ICancelable? Timer;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Initializes the instance actor with its configuration and dependencies.
|
||||
@@ -388,10 +406,10 @@ public class InstanceActor : ReceiveActor
|
||||
/// <inheritdoc />
|
||||
protected override void PostStop()
|
||||
{
|
||||
// Cancel any armed tag-subscribe retry timers so they cannot fire after stop.
|
||||
foreach (var timer in _tagSubscribeRetryTimers.Values)
|
||||
timer.Cancel();
|
||||
_tagSubscribeRetryTimers.Clear();
|
||||
// Cancel any armed tag-subscribe watchdogs so they cannot fire after stop.
|
||||
foreach (var state in _tagSubscribeStates.Values)
|
||||
state.Timer?.Cancel();
|
||||
_tagSubscribeStates.Clear();
|
||||
|
||||
// Final flush of any coalesced static-override writes so a graceful stop does not lose
|
||||
// the last <= flush-interval of writes (P4 pattern, mirrors NativeAlarmActor.PostStop).
|
||||
@@ -1153,6 +1171,24 @@ public class InstanceActor : ReceiveActor
|
||||
_logger.LogWarning("Connection {Connection} quality changed to {Quality} for instance {Instance}",
|
||||
qualityChanged.ConnectionName, qualityChanged.Quality, _instanceUniqueName);
|
||||
|
||||
// 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. Timer != null ⇔ the
|
||||
// subscribe is unconfirmed (it is only nulled by a Success); if a request is in
|
||||
// flight and young the re-armed tick just hits the guard and re-arms at the floor.
|
||||
// Only acts while the interval is actually stretched: once reset to the floor a
|
||||
// further Good is a no-op, so a flapping connection cannot keep pushing the armed
|
||||
// tick out and starve the retry it was meant to hasten.
|
||||
if (qualityChanged.Quality == QualityCode.Good
|
||||
&& _tagSubscribeStates.TryGetValue(qualityChanged.ConnectionName, out var subState)
|
||||
&& subState.Timer != null
|
||||
&& subState.Interval > TimeSpan.FromMilliseconds(_options.TagSubscribeRetryIntervalMs))
|
||||
{
|
||||
ResetTagSubscribeBackoff(subState);
|
||||
ArmTagSubscribeWatchdog(qualityChanged.ConnectionName, subState);
|
||||
}
|
||||
|
||||
if (_configuration == null) return;
|
||||
|
||||
// Mark all attributes bound to this connection with the new quality
|
||||
@@ -1190,9 +1226,21 @@ public class InstanceActor : ReceiveActor
|
||||
{
|
||||
if (_dclManager == null || _configuration == null) return;
|
||||
|
||||
// Defensive — PreStart is the only caller today.
|
||||
// Lever 1: prune retry state from a previous build — cancels any armed watchdog and
|
||||
// drops its pending row so a re-entry can never strand an in-flight slot.
|
||||
foreach (var state in _tagSubscribeStates.Values)
|
||||
{
|
||||
state.Timer?.Cancel();
|
||||
if (state.InFlightCorrelationId != null)
|
||||
_pendingTagSubscribes.Remove(state.InFlightCorrelationId);
|
||||
}
|
||||
_tagSubscribeStates.Clear();
|
||||
|
||||
// Group attributes by their bound connection name into the persistent
|
||||
// _tagsByConnection field so a retry can re-send the exact same request.
|
||||
_tagsByConnection.Clear();
|
||||
_tagPathToAttributes.Clear();
|
||||
foreach (var attr in _configuration.Attributes)
|
||||
{
|
||||
if (string.IsNullOrEmpty(attr.DataSourceReference) ||
|
||||
@@ -1224,12 +1272,39 @@ public class InstanceActor : ReceiveActor
|
||||
SendTagSubscribe(connectionName);
|
||||
}
|
||||
|
||||
/// <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(Math.Round(clamped));
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Sends (or re-sends) the tag-subscribe request for one connection and arms a
|
||||
/// per-connection retry timer (S4/UA6). Arming on send means the retry fires
|
||||
/// whether the response is a failure OR is lost entirely; a Success reply
|
||||
/// cancels it. A superseded in-flight correlation id is overwritten so only the
|
||||
/// latest attempt's reply is honoured.
|
||||
/// per-connection watchdog (S4/UA6). Arming on send means the retry fires whether
|
||||
/// the response is a failure OR is lost entirely; a Success reply cancels it.
|
||||
/// <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)
|
||||
{
|
||||
@@ -1237,51 +1312,123 @@ public class InstanceActor : ReceiveActor
|
||||
if (!_tagsByConnection.TryGetValue(connectionName, out var tagPaths) || tagPaths.Count == 0)
|
||||
return;
|
||||
|
||||
var state = GetTagSubscribeState(connectionName);
|
||||
var now = Environment.TickCount64;
|
||||
|
||||
if (state.InFlightCorrelationId != null)
|
||||
{
|
||||
var ageMs = now - state.InFlightSentTicks;
|
||||
if (ageMs < _options.TagSubscribeInFlightCeilingMs)
|
||||
{
|
||||
_logger.LogDebug(
|
||||
"Instance {Instance} tag subscribe on connection {Connection} still in flight ({AgeMs}ms, corr={Corr}) — not re-sending",
|
||||
_instanceUniqueName, connectionName, ageMs, state.InFlightCorrelationId);
|
||||
ArmTagSubscribeWatchdog(connectionName, state);
|
||||
return;
|
||||
}
|
||||
|
||||
_logger.LogWarning(
|
||||
"Instance {Instance} tag subscribe on connection {Connection} unanswered for {AgeMs}ms (corr={Corr}) — presuming lost, re-sending",
|
||||
_instanceUniqueName, connectionName, ageMs, state.InFlightCorrelationId);
|
||||
_pendingTagSubscribes.Remove(state.InFlightCorrelationId);
|
||||
BackOffTagSubscribe(state);
|
||||
}
|
||||
|
||||
var correlationId = Guid.NewGuid().ToString("N");
|
||||
_pendingTagSubscribes[correlationId] = connectionName;
|
||||
state.InFlightCorrelationId = correlationId;
|
||||
state.InFlightSentTicks = now;
|
||||
|
||||
var request = new SubscribeTagsRequest(
|
||||
correlationId, _instanceUniqueName, connectionName, tagPaths, DateTimeOffset.UtcNow);
|
||||
_dclManager.Tell(request, Self);
|
||||
_logger.LogInformation(
|
||||
"Instance {Instance} subscribing to {Count} tags on connection {Connection} (corr={Corr})",
|
||||
_instanceUniqueName, tagPaths.Count, connectionName, correlationId);
|
||||
"Instance {Instance} subscribing to {Count} tags on connection {Connection} (corr={Corr}, next retry in {Interval})",
|
||||
_instanceUniqueName, tagPaths.Count, connectionName, correlationId, state.Interval);
|
||||
|
||||
if (_tagSubscribeRetryTimers.Remove(connectionName, out var existing))
|
||||
existing.Cancel();
|
||||
_tagSubscribeRetryTimers[connectionName] = Context.System.Scheduler.ScheduleTellOnceCancelable(
|
||||
TimeSpan.FromMilliseconds(_options.TagSubscribeRetryIntervalMs), Self,
|
||||
new RetryTagSubscribe(connectionName), Self);
|
||||
ArmTagSubscribeWatchdog(connectionName, state);
|
||||
}
|
||||
|
||||
private TagSubscribeState GetTagSubscribeState(string connectionName)
|
||||
{
|
||||
if (!_tagSubscribeStates.TryGetValue(connectionName, out var state))
|
||||
{
|
||||
state = new TagSubscribeState { Interval = TimeSpan.FromMilliseconds(_options.TagSubscribeRetryIntervalMs) };
|
||||
_tagSubscribeStates[connectionName] = state;
|
||||
}
|
||||
return state;
|
||||
}
|
||||
|
||||
/// <summary>(Re)arms the per-connection watchdog at the state's current interval.</summary>
|
||||
private void ArmTagSubscribeWatchdog(string connectionName, TagSubscribeState state)
|
||||
{
|
||||
state.Timer?.Cancel();
|
||||
state.Timer = Context.System.Scheduler.ScheduleTellOnceCancelable(
|
||||
state.Interval, Self, new RetryTagSubscribe(connectionName), Self);
|
||||
}
|
||||
|
||||
/// <summary>Advances the backoff after a failed or presumed-lost attempt.</summary>
|
||||
private void BackOffTagSubscribe(TagSubscribeState state) =>
|
||||
state.Interval = NextTagSubscribeInterval(
|
||||
state.Interval,
|
||||
TimeSpan.FromMilliseconds(_options.TagSubscribeRetryIntervalMs),
|
||||
TimeSpan.FromMilliseconds(_options.TagSubscribeRetryMaxIntervalMs),
|
||||
Random.Shared.NextDouble());
|
||||
|
||||
/// <summary>Resets the backoff to its floor (Success, or connection back to Good).</summary>
|
||||
private void ResetTagSubscribeBackoff(TagSubscribeState state) =>
|
||||
state.Interval = TimeSpan.FromMilliseconds(_options.TagSubscribeRetryIntervalMs);
|
||||
|
||||
/// <summary>
|
||||
/// Handles the DCL's tag-subscribe response: on Success cancels the connection's
|
||||
/// retry timer; on failure logs + emits a site event and leaves the armed timer
|
||||
/// to re-send. A reply for an unknown/superseded correlation id is ignored.
|
||||
/// Handles the DCL's tag-subscribe response. A reply whose correlation id is not the
|
||||
/// pending one (superseded at the in-flight ceiling, or a duplicate) is ignored. On
|
||||
/// Success the connection's watchdog is cancelled, the in-flight slot cleared and the
|
||||
/// backoff reset to the floor. On failure (Lever 1, 2026-08-28) the slot is freed, the
|
||||
/// backoff advanced (×2 ±20 % jitter, capped) and the watchdog re-armed at the new
|
||||
/// interval, so the next attempt waits the backed-off interval — not the in-flight
|
||||
/// ceiling — and duplicates are still impossible.
|
||||
/// </summary>
|
||||
private void HandleSubscribeTagsResponse(SubscribeTagsResponse response)
|
||||
{
|
||||
if (!_pendingTagSubscribes.Remove(response.CorrelationId, out var connectionName))
|
||||
return; // stale/superseded reply — ignore
|
||||
|
||||
// Read, never create: the id was minted by SendTagSubscribe, which creates the
|
||||
// state, so a missing entry means the retry state was pruned (SubscribeToDcl
|
||||
// re-entry) and this reply no longer governs anything.
|
||||
if (!_tagSubscribeStates.TryGetValue(connectionName, out var state))
|
||||
return;
|
||||
|
||||
if (response.Success)
|
||||
{
|
||||
if (_tagSubscribeRetryTimers.Remove(connectionName, out var timer))
|
||||
timer.Cancel();
|
||||
if (state.InFlightCorrelationId == response.CorrelationId)
|
||||
{
|
||||
state.Timer?.Cancel();
|
||||
state.Timer = null;
|
||||
state.InFlightCorrelationId = null;
|
||||
ResetTagSubscribeBackoff(state);
|
||||
}
|
||||
_logger.LogDebug(
|
||||
"Instance {Instance} tag subscription confirmed on connection {Connection}",
|
||||
_instanceUniqueName, connectionName);
|
||||
}
|
||||
else
|
||||
{
|
||||
if (state.InFlightCorrelationId == response.CorrelationId)
|
||||
{
|
||||
// A definite failure: the slot is free, back off, and re-arm at the NEW
|
||||
// interval so the backoff is honoured on this very retry, not the one after.
|
||||
state.InFlightCorrelationId = null;
|
||||
BackOffTagSubscribe(state);
|
||||
ArmTagSubscribeWatchdog(connectionName, state);
|
||||
}
|
||||
_logger.LogWarning(
|
||||
"Instance {Instance} tag subscribe FAILED on connection {Connection}: {Error} — will retry",
|
||||
_instanceUniqueName, connectionName, response.ErrorMessage);
|
||||
"Instance {Instance} tag subscribe FAILED on connection {Connection}: {Error} — will retry in {Interval}",
|
||||
_instanceUniqueName, connectionName, response.ErrorMessage, state.Interval);
|
||||
_ = _serviceProvider?.GetService<ISiteEventLogger>()?.LogEventAsync(
|
||||
"instance_lifecycle", "Warning", _instanceUniqueName,
|
||||
$"InstanceActor:{_instanceUniqueName}",
|
||||
$"Tag subscribe failed on connection '{connectionName}': {response.ErrorMessage} — will retry");
|
||||
// Leave the armed retry timer to re-send.
|
||||
$"Tag subscribe failed on connection '{connectionName}': {response.ErrorMessage} — will retry in {state.Interval}");
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -112,13 +112,43 @@ public class SiteRuntimeOptions
|
||||
public int ConfigFetchTimeoutSeconds { get; set; } = 30;
|
||||
|
||||
/// <summary>
|
||||
/// Fixed 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, so it closes
|
||||
/// both the failed-response and the lost-response gaps. Default: 5000ms.
|
||||
/// 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.
|
||||
/// <para>
|
||||
/// Two consequences worth knowing: (1) the guard binds only while the retry interval
|
||||
/// is shorter than this ceiling — once the backoff has grown past it every tick is
|
||||
/// already older than the ceiling, so the interval itself becomes the rate limit
|
||||
/// (steady state at the defaults: one re-send per 60 s); (2) a connection returning to
|
||||
/// Good resets the backoff but does NOT release an outstanding request — that request
|
||||
/// still owns the connection until the ceiling, so the Good reset shortens the wait
|
||||
/// only when nothing is in flight.
|
||||
/// </para>
|
||||
/// </summary>
|
||||
public int TagSubscribeInFlightCeilingMs { get; set; } = 45000;
|
||||
|
||||
/// <summary>
|
||||
/// Grace period (ms) after a script's execution timeout elapses before the
|
||||
/// stuck-script watchdog declares the script's dedicated thread blocked and
|
||||
|
||||
@@ -79,6 +79,16 @@ public sealed class SiteRuntimeOptionsValidator : OptionsValidatorBase<SiteRunti
|
||||
$"(was {options.TagSubscribeRetryIntervalMs}); a zero interval hot-loops the tag-subscribe " +
|
||||
"retry and a negative one throws inside the Instance Actor's retry scheduling.");
|
||||
|
||||
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.");
|
||||
|
||||
builder.RequireThat(options.StuckScriptGraceMs >= 0,
|
||||
$"ScadaBridge:SiteRuntime:StuckScriptGraceMs must be >= 0 " +
|
||||
$"(was {options.StuckScriptGraceMs}); a negative grace throws inside the stuck-script watchdog's delay.");
|
||||
|
||||
+51
@@ -0,0 +1,51 @@
|
||||
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);
|
||||
}
|
||||
}
|
||||
@@ -1290,7 +1290,7 @@ public class InstanceActorTests : TestKit, IDisposable
|
||||
[Fact]
|
||||
public void FailedSubscribeResponse_IsRetried()
|
||||
{
|
||||
var options = new SiteRuntimeOptions { TagSubscribeRetryIntervalMs = 200 };
|
||||
var options = new SiteRuntimeOptions { TagSubscribeRetryIntervalMs = 200, TagSubscribeInFlightCeilingMs = 60_000 };
|
||||
var (instance, dclProbe) = CreateInstanceWithDataSourcedAttribute(options);
|
||||
|
||||
var first = dclProbe.ExpectMsg<SubscribeTagsRequest>();
|
||||
@@ -1305,7 +1305,8 @@ public class InstanceActorTests : TestKit, IDisposable
|
||||
[Fact]
|
||||
public void LostSubscribeResponse_IsRetried_AndSuccessStopsRetrying()
|
||||
{
|
||||
var options = new SiteRuntimeOptions { TagSubscribeRetryIntervalMs = 200 };
|
||||
// Lever 1: a lost reply is re-sent only once the outstanding request is presumed lost (ceiling), by design.
|
||||
var options = new SiteRuntimeOptions { TagSubscribeRetryIntervalMs = 200, TagSubscribeInFlightCeilingMs = 200 };
|
||||
var (instance, dclProbe) = CreateInstanceWithDataSourcedAttribute(options);
|
||||
|
||||
dclProbe.ExpectMsg<SubscribeTagsRequest>();
|
||||
@@ -1316,4 +1317,223 @@ public class InstanceActorTests : TestKit, IDisposable
|
||||
true, null, DateTimeOffset.UtcNow), dclProbe.Ref);
|
||||
dclProbe.ExpectNoMsg(TimeSpan.FromMilliseconds(600)); // retry timer cancelled
|
||||
}
|
||||
|
||||
// --- Lever 1 (2026-08-28): in-flight guard ---
|
||||
|
||||
[Fact]
|
||||
public void SlowSubscribe_DoesNotStackDuplicates_WhileInFlight()
|
||||
{
|
||||
// Retry ticks every 200 ms, but an unanswered request blocks a duplicate for 2 s.
|
||||
var options = new SiteRuntimeOptions
|
||||
{
|
||||
TagSubscribeRetryIntervalMs = 200,
|
||||
TagSubscribeRetryMaxIntervalMs = 200,
|
||||
TagSubscribeInFlightCeilingMs = 2000,
|
||||
};
|
||||
var (_, dclProbe) = CreateInstanceWithDataSourcedAttribute(options);
|
||||
|
||||
dclProbe.ExpectMsg<SubscribeTagsRequest>();
|
||||
// Pre-Lever-1 this window would carry ~4 more requests (one per 200 ms tick).
|
||||
dclProbe.ExpectNoMsg(TimeSpan.FromMilliseconds(1000));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void SlowSubscribe_ReissuesAfterInFlightCeiling_AndOnlyOnce()
|
||||
{
|
||||
var options = new SiteRuntimeOptions
|
||||
{
|
||||
TagSubscribeRetryIntervalMs = 200,
|
||||
TagSubscribeRetryMaxIntervalMs = 200,
|
||||
TagSubscribeInFlightCeilingMs = 600,
|
||||
};
|
||||
var (_, dclProbe) = CreateInstanceWithDataSourcedAttribute(options);
|
||||
|
||||
var first = dclProbe.ExpectMsg<SubscribeTagsRequest>();
|
||||
var second = dclProbe.ExpectMsg<SubscribeTagsRequest>(TimeSpan.FromSeconds(3)); // after ceiling
|
||||
Assert.NotEqual(first.CorrelationId, second.CorrelationId);
|
||||
Assert.Equal(first.TagPaths, second.TagPaths);
|
||||
// The new request is itself guarded — no third one inside its ceiling window.
|
||||
dclProbe.ExpectNoMsg(TimeSpan.FromMilliseconds(400));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void ReplyToSupersededRequest_IsIgnored_AndLatestStillGoverns()
|
||||
{
|
||||
var options = new SiteRuntimeOptions
|
||||
{
|
||||
TagSubscribeRetryIntervalMs = 200,
|
||||
TagSubscribeRetryMaxIntervalMs = 200,
|
||||
TagSubscribeInFlightCeilingMs = 300,
|
||||
};
|
||||
var (instance, dclProbe) = CreateInstanceWithDataSourcedAttribute(options);
|
||||
|
||||
var first = dclProbe.ExpectMsg<SubscribeTagsRequest>();
|
||||
var second = dclProbe.ExpectMsg<SubscribeTagsRequest>(TimeSpan.FromSeconds(3));
|
||||
|
||||
// A late Success for the FIRST id is dropped at the pending-map check (its id was
|
||||
// removed at the ceiling); the watchdog guarding the SECOND is untouched.
|
||||
instance.Tell(new SubscribeTagsResponse(first.CorrelationId, first.InstanceUniqueName,
|
||||
true, null, DateTimeOffset.UtcNow), dclProbe.Ref);
|
||||
var third = dclProbe.ExpectMsg<SubscribeTagsRequest>(TimeSpan.FromSeconds(3)); // second eventually re-sent
|
||||
Assert.NotEqual(second.CorrelationId, third.CorrelationId);
|
||||
|
||||
// Success for the live id stops everything.
|
||||
instance.Tell(new SubscribeTagsResponse(third.CorrelationId, third.InstanceUniqueName,
|
||||
true, null, DateTimeOffset.UtcNow), dclProbe.Ref);
|
||||
dclProbe.ExpectNoMsg(TimeSpan.FromMilliseconds(800));
|
||||
}
|
||||
|
||||
// --- 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);
|
||||
// gaps[1] compounds two jitter draws: 2×(2×200×0.8)×0.8 = 512 ms minimum.
|
||||
Assert.InRange(gaps[1].TotalMilliseconds, 500, 1100);
|
||||
Assert.InRange(gaps[2].TotalMilliseconds, 600, 1100);
|
||||
|
||||
// Success resets: no further request follows a confirmed subscription.
|
||||
instance.Tell(new SubscribeTagsResponse(req.CorrelationId, req.InstanceUniqueName,
|
||||
true, null, DateTimeOffset.UtcNow), dclProbe.Ref);
|
||||
dclProbe.ExpectNoMsg(TimeSpan.FromMilliseconds(1200));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void FailureReply_ReissuesEvenThoughGuardHasNotExpired()
|
||||
{
|
||||
var options = new SiteRuntimeOptions
|
||||
{
|
||||
TagSubscribeRetryIntervalMs = 200,
|
||||
TagSubscribeRetryMaxIntervalMs = 200,
|
||||
TagSubscribeInFlightCeilingMs = 60_000,
|
||||
};
|
||||
var (instance, dclProbe) = CreateInstanceWithDataSourcedAttribute(options);
|
||||
|
||||
var first = dclProbe.ExpectMsg<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);
|
||||
}
|
||||
|
||||
// --- Lever 1 (2026-08-28): Good quality resets the backoff ---
|
||||
|
||||
[Fact]
|
||||
public void FlappingGood_DoesNotDeferTheRetry()
|
||||
{
|
||||
// A burst of Good events faster than the floor must not keep re-arming the watchdog
|
||||
// from "now" — after the first reset the interval is at the floor and later Goods are
|
||||
// no-ops, so the retry still lands within one floor interval of the first Good.
|
||||
var options = new SiteRuntimeOptions
|
||||
{
|
||||
TagSubscribeRetryIntervalMs = 300,
|
||||
TagSubscribeRetryMaxIntervalMs = 5000,
|
||||
TagSubscribeInFlightCeilingMs = 60_000,
|
||||
};
|
||||
var (instance, dclProbe) = CreateInstanceWithDataSourcedAttribute(options);
|
||||
|
||||
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));
|
||||
}
|
||||
instance.Tell(new SubscribeTagsResponse(req.CorrelationId, req.InstanceUniqueName,
|
||||
false, "busy", DateTimeOffset.UtcNow), dclProbe.Ref);
|
||||
|
||||
// Twelve Goods over ~1.2 s — a naive re-arm-from-now on each would push the tick to ≥1.5 s.
|
||||
for (var i = 0; i < 12; i++)
|
||||
{
|
||||
instance.Tell(new ConnectionQualityChanged("conn-1", QualityCode.Good, DateTimeOffset.UtcNow));
|
||||
Thread.Sleep(100);
|
||||
}
|
||||
|
||||
// The retry must have landed DURING the burst (~300 ms after the first Good), so the probe
|
||||
// already holds it — assert that directly rather than timing the burst's Thread.Sleeps.
|
||||
dclProbe.ExpectMsg<SubscribeTagsRequest>(TimeSpan.FromMilliseconds(50));
|
||||
}
|
||||
|
||||
[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");
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void LateReplyForFailedAndReissuedRequest_IsIgnored()
|
||||
{
|
||||
var options = new SiteRuntimeOptions
|
||||
{
|
||||
TagSubscribeRetryIntervalMs = 200,
|
||||
TagSubscribeRetryMaxIntervalMs = 200,
|
||||
// Short ceiling on purpose: the second request stays in flight (nothing ever
|
||||
// answers it), so the in-flight guard is what re-sends. With a 60 s ceiling the
|
||||
// guard would hold for 60 s and the test could not tell "late reply ignored"
|
||||
// apart from "late reply wrongly cleared the slot" — both look like silence.
|
||||
TagSubscribeInFlightCeilingMs = 500,
|
||||
};
|
||||
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 once its
|
||||
// in-flight ceiling expires. Had the late reply been honoured it would have cancelled
|
||||
// that watchdog and cleared the slot, and no third request would ever be sent.
|
||||
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);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -144,4 +144,38 @@ public class SiteRuntimeOptionsValidatorTests
|
||||
Assert.True(result.Failed);
|
||||
Assert.Contains("MaxConcurrentRunsPerScript", result.FailureMessage);
|
||||
}
|
||||
// --- 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);
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user