Compare commits
6 Commits
d3be5f0527
...
d18f344e66
| Author | SHA1 | Date | |
|---|---|---|---|
 Joseph Doherty
|
d18f344e66 | ||
|
Joseph Doherty
|
042e2efd1d | ||
|
Joseph Doherty
|
61a14bccab | ||
|
Joseph Doherty
|
59468399db | ||
|
Joseph Doherty
|
c4d3c8b875 | ||
|
Joseph Doherty
|
e9e44a9ef0 |
@@ -99,7 +99,7 @@ spec for each is `docs/requirements/Component-<Name>.md`, and `README.md` carrie
|
||||
- **`ActiveNodeEvaluator.SelfIsOldestUp` is THE single definition of "active node"** (`Communication/ClusterState/ActiveNodeEvaluator.cs`) — the **oldest Up member** in a role scope, and explicitly **never `cluster.State.Leader`**: leadership (lowest address) is an Akka-internal concept that diverges from singleton placement permanently once the original first node restarts and rejoins, and both sides claim it during a partition. The equivalence *oldest-Up == where `ClusterSingletonManager` places singletons* **is** the design. `ClusterActivityEvaluator.SelfIsOldest`, the S&F delivery gate, `/health/active` and the heartbeat `IsActive` stamp all delegate here.
|
||||
- Site nodes carry **two Akka roles**: the base `Site` plus a site-specific `site-{SiteId}` (`AkkaHostedService.BuildRoles`). Singletons scope to the **site-specific** role.
|
||||
- **The gRPC boundary is authenticated (PSK) as of 2026-07-22; Akka remoting still is not, and nothing is encrypted.** Akka remoting sets no `enable-ssl`, no secure cookie, no `trusted-selection-paths` — so intra-cluster Akka remoting remains open to anyone who can reach the remoting port, and that boundary still assumes a trusted network. The gRPC listener stays **h2c**, but `SiteStreamService` is no longer open: `ControlPlaneAuthInterceptor` (`Host/ControlPlaneAuthInterceptor.cs`) gates `/sitestream.SiteStreamService/` — including the `PullAuditEvents`/`PullSiteCalls` RPCs that return audit rows — against a **per-site preshared key**, fail-closed, constant-time compared, alongside the separate `LocalDbSyncAuthInterceptor` on `/localdb_sync.v1.LocalDbSync/` with its own separate key. **Site side:** `ScadaBridge:Communication:GrpcPsk`, in production `${secret:SB-GRPC-PSK-<siteId>}`, and **`StartupValidator` refuses to boot a site node without it** (an unset key would leave the node healthy-looking but serving nothing). **Central side:** `SitePskProvider` resolves `SB-GRPC-PSK-{siteId}` from the secrets store at channel-build time (sites are added at runtime, so no boot-time expansion is possible), with `ScadaBridge:Communication:SitePsks:{siteId}` as an override for hosts running without a master key — the docker rig uses the latter. One key per site, never fleet-wide. A bearer token over h2c is readable and replayable on-path; TLS is the follow-on hardening and needs no change to this design. Introduced by Phase 0 of the ClusterClient→gRPC migration (`docs/plans/2026-07-22-clusterclient-to-grpc-plan.md`).
|
||||
- gRPC streaming channel — **note the direction is inverted from the data flow**: data moves site→central, but each **site node hosts the gRPC server** (`SiteStreamGrpcServer`, Kestrel h2c, port 8083, mapped **only in the Site branch** of `Program.cs`) and **central is the client**, dialling in. Central creates per-site `SiteStreamGrpcClient` via `SiteStreamGrpcClientFactory`, keyed **`(siteId, endpoint)`** — the key was widened from site-only to fix an arch-review High where one session's NodeA→NodeB flip disposed a channel another session was still using. Proto evolution is **additive only** and field numbers are never reused (`AlarmStateUpdate` grew 7→23 fields for the native-alarm mirror, then 24 for `AckTime`). Generated C# is **vendored** under `Communication/SiteStreamGrpc/` with the `<Protobuf>` include commented out — regeneration is a manual toggle-build-copy-untoggle, automated by `docker/regen-proto.sh [sitestream|centralcontrol|sitecommand|all]` (it always restores the csproj; an active `<Protobuf>` item must never be committed — it breaks the Docker build). **Stream events are BATCHED as of R2 (2026-08-15)** — a site emitting ~37.5k events/s used to pay one gRPC message per event. Additive wire shape, no new RPC: `batching_supported` on the request (`InstanceStreamRequest` field 3 / `SiteStreamRequest` field 2) plus a `SiteStreamEvent.batch` oneof case (**field 4**) carrying a new `SiteStreamEventBatch { repeated SiteStreamEvent events = 1 }`. **The proto3 default of that flag IS the negotiation, and it is load-bearing** — a batch frame reaches a pre-R2 central as `EventOneofCase.None`, whose `ConvertToDomainEvent` returns null, so the whole batch would vanish with no error anywhere; an old central cannot set the flag, so it never gets one, and an old site ignores the unknown request field and keeps sending per-event frames the new client's `ForEachEvent` unpack handles as the single-event case. Server side is `SiteStreamEventBatcher`, a per-subscriber pump **downstream of `StreamRelayActor`'s bounded DropOldest channel** — so it changes framing only and **does NOT move the burst ceiling** (deferred register row 31; that ceiling lives in the shared publish stage upstream of the BroadcastHub). It never delays a lone event: it drains the already-queued backlog for free and lingers only once a backlog is proven, then emits a single-event buffer as a plain frame. `GrpcStreamBatchMaxEvents` (100, `1` disables) and `GrpcStreamBatchWindow` (25 ms, validated **strictly under 250 ms** = the load test's end-to-end P99 threshold); measured worst case P99 25.4 ms. Histogram `scadabridge.site.stream.batch_size`, recorded only on negotiated streams (`ScadaBridgeTelemetry.MeterName` is already in the `ObservedMeters` allowlist). **The client reconnects on graceful (OK-status) stream completion, not just on fault (arch-review remediation, commit `34a3f4bb`)** — Kestrel's `MaxStreamLifetime`/`Grpc.AspNetCore.Server` max-connection-age periodically ends a healthy stream with a normal completion, which the client used to treat as terminal (no reconnect attempt), silently killing a site's live feed until the next process restart (observed up to ~4h on the rig); live-probed at a 2-minute forced lifetime, reconnect lands within one reconcile tick and `IsLive` reflects the gap in between.
|
||||
- gRPC streaming channel — **note the direction is inverted from the data flow**: data moves site→central, but each **site node hosts the gRPC server** (`SiteStreamGrpcServer`, Kestrel h2c, port 8083, mapped **only in the Site branch** of `Program.cs`) and **central is the client**, dialling in. Central creates per-site `SiteStreamGrpcClient` via `SiteStreamGrpcClientFactory`, keyed **`(siteId, endpoint)`** — the key was widened from site-only to fix an arch-review High where one session's NodeA→NodeB flip disposed a channel another session was still using. Proto evolution is **additive only** and field numbers are never reused (`AlarmStateUpdate` grew 7→23 fields for the native-alarm mirror, then 24 for `AckTime`). Generated C# is **vendored** under `Communication/SiteStreamGrpc/` with the `<Protobuf>` include commented out — regeneration is a manual toggle-build-copy-untoggle, automated by `docker/regen-proto.sh [sitestream|centralcontrol|sitecommand|all]` (it always restores the csproj; an active `<Protobuf>` item must never be committed — it breaks the Docker build). **Stream events are BATCHED as of R2 (2026-08-15)** — a site emitting ~37.5k events/s used to pay one gRPC message per event. Additive wire shape, no new RPC: `batching_supported` on the request (`InstanceStreamRequest` field 3 / `SiteStreamRequest` field 2) plus a `SiteStreamEvent.batch` oneof case (**field 4**) carrying a new `SiteStreamEventBatch { repeated SiteStreamEvent events = 1 }`. **The proto3 default of that flag IS the negotiation, and it is load-bearing** — a batch frame reaches a pre-R2 central as `EventOneofCase.None`, whose `ConvertToDomainEvent` returns null, so the whole batch would vanish with no error anywhere; an old central cannot set the flag, so it never gets one, and an old site ignores the unknown request field and keeps sending per-event frames the new client's `ForEachEvent` unpack handles as the single-event case. Server side is `SiteStreamEventBatcher`, a per-subscriber pump **downstream of `StreamRelayActor`'s bounded DropOldest channel** — so it changes framing only and **does NOT move the burst ceiling** (deferred register row 31; that ceiling lives in the shared publish stage upstream of the BroadcastHub). It never delays a lone event: it drains the already-queued backlog for free and lingers only once a backlog is proven, then emits a single-event buffer as a plain frame. `GrpcStreamBatchMaxEvents` (100, `1` disables) and `GrpcStreamBatchWindow` (25 ms, validated **strictly under 250 ms** = the load test's end-to-end P99 threshold); measured worst case P99 25.4 ms. **Benchmark-proven at target scale 2026-08-18** (the LoadHarness wire path — real Kestrel h2c + real `SiteStreamGrpcClient`, `--wire-probes-per-site`): 899,807 events in 12,462 frames (mean 72.2 ev/frame), wire P99 43.4 ms (5.8× inside the 250 ms budget — a 75-event burst never hits the 100 cap, so every batch pays the full 25 ms window: this workload's worst-case latency posture), zero drops/faults; results doc §8. Histogram `scadabridge.site.stream.batch_size`, recorded only on negotiated streams (`ScadaBridgeTelemetry.MeterName` is already in the `ObservedMeters` allowlist). **The client reconnects on graceful (OK-status) stream completion, not just on fault (arch-review remediation, commit `34a3f4bb`)** — Kestrel's `MaxStreamLifetime`/`Grpc.AspNetCore.Server` max-connection-age periodically ends a healthy stream with a normal completion, which the client used to treat as terminal (no reconnect attempt), silently killing a site's live feed until the next process restart (observed up to ~4h on the rig); live-probed at a 2-minute forced lifetime, reconnect lands within one reconcile tick and `IsLive` reflects the gap in between.
|
||||
- Native alarms are a **read-only** mirror of OPC UA Alarms & Conditions and MxAccess Gateway alarms — **no ack-back, no central tables**; state lives in the site's `native_alarm_state`, survives failover, and is cleared on redeploy/undeploy (mirrors static overrides). Central's per-site live alarm cache (`ISiteAlarmLiveCache`) is **transient in-memory only** — there is deliberately no persisted central alarm store, so the 15s poll remains the NotReporting authority behind the live stream. See `Component-DataConnectionLayer.md` / `Component-CentralUI.md` for the model and the authoring surface.
|
||||
- **`AckTime` mirror enrichment + the `Alarms` script accessor (MES alarm-status API Phase 1, 2026-08-01).** `AlarmStateChanged` carries an additive `AckTime` (`DateTimeOffset?`), mirrored on the vendored `AlarmStateUpdate` proto as **field 24** and persisted inside `native_alarm_state`'s `metadata_json` — deliberately NOT a new column, because that table is `RegisterReplicated` and LocalDb builds its CDC triggers from the column list at registration time. Set only while a condition is active AND acknowledged (so it is null while unacked and cleared on re-raise); the DCL stamps the source's own ack instant for OPC UA (new SelectClause **index 18** = `AckedState/TransitionTime`) and its observation time of the ack transition for MxGateway, which supplies none. Site `Call` scripts read alarms via the new **`Alarms.CurrentAsync()`** accessor (`ScriptRuntimeContext` + `ScriptGlobals`, local Ask on `GetAlarmSnapshotRequest`, returns `Commons.Types.Scripts.ScriptAlarm`), mirrored on `ScriptCompileSurface` AND the Central UI `SandboxScriptHost` editor surface. The trust model needed no change — it is a deny-list over API roots, not an allow-list of context members. Plan: `docs/plans/2026-06-30-mes-alarm-status-api.md` (Phases 2–4 are deployed config, not repo).
|
||||
- OPC UA cert trust is **site-local and not persisted centrally** (follow-up): the verify-endpoint probe captures an untrusted server cert but **NEVER trusts it**, and DeploymentManager broadcasts `TrustServerCertCommand`/`RemoveServerCertCommand` to **BOTH** site nodes — `CertStoreActor` runs on every site node, not as a singleton, so PKI stores stay consistent across failover.
|
||||
|
||||
@@ -191,7 +191,7 @@ Each is a deliberate, recorded trade — not an omission.
|
||||
| D1 | **Sustained window shortened from 1 hour to 20 minutes** at full scale | Practicality on a single shared workstation. Memory growth is reported as a *slope* precisely so a shorter window still answers the leak question: a leak shows as a positive slope in 20 min just as in 60. The 1-hour run is a single flag — `--sustain-minutes 60` — and the exact command is recorded in the results doc. |
|
||||
| D2 | **In-process, non-clustered sites** rather than 10 real two-node Akka clusters | A single box cannot host 20 clustered nodes at this scale. Cluster membership/failover is already measured on a real two-node rig (`FailoverTimingTests`, `docker/failover-drill.sh`); WP-4's question is about the hierarchy under the singleton. |
|
||||
| D3 | **Simulated data source** instead of real OPC UA | No simulated adapter exists in the DCL, and 375k real monitored items would measure the OPC UA SDK. Injected at the documented `RegisterAdapter` seam, at the exact hand-off point the real adapter uses. |
|
||||
| D4 | **gRPC socket replaced by a reader task** on the stream path | Required by the row-50 measurement itself: a stalled subscriber must be *held* stalled. All of `SiteStreamGrpcServer`'s per-subscription machinery above the socket is real. |
|
||||
| D4 | **gRPC socket replaced by a reader task** on the stream path | Required by the row-50 measurement itself: a stalled subscriber must be *held* stalled. All of `SiteStreamGrpcServer`'s per-subscription machinery above the socket is real. **Partially retired 2026-08-18:** the harness now ALSO attaches per-site wire probes — real Kestrel h2c + real `SiteStreamGrpcClient` with R2 batching negotiated (`--wire-probes-per-site`, default 10) — so the batcher/serialization/HTTP/2 hop is measured too; the substituted-reader probes remain because the slow-subscriber scenario still needs them. Results doc §8. |
|
||||
| D5 | **Instance Actors are `/user` children**, not children of a `DeploymentManagerActor` | Constructing them directly is what lets the ramp be timed in isolation and keeps the harness independent of the deploy round-trip. Visible only as `InstanceActorInitialized` dead letters, which the harness suppresses. No measured path differs. |
|
||||
| D6 | **Akka logging at WARNING** | At 37,500 updates/s, INFO output would itself become a measured load. |
|
||||
| D7 | `[xc-3]` site event logging volume, `[xc-4]` audit-log degradation, `[xc-5]` template flattening, `[xc-8]` UI responsiveness **not covered** | Out of scope for this harness: the first three need central MS SQL and the Template Engine (a central-cluster fixture, not a site one), and `[xc-8]` needs a browser. Recorded as residual scope in the results doc rather than silently dropped. |
|
||||
|
||||
@@ -275,7 +275,10 @@ Stated plainly so the evidence is not over-read:
|
||||
- **Not a real-network run.** The gRPC, health-transport and S&F-to-central hops are
|
||||
in-process. Serialization cost, TLS, WAN latency and socket backpressure are out of
|
||||
frame; the *stream* path's backpressure was measured deliberately by substituting a
|
||||
controllable reader for the socket (D4).
|
||||
controllable reader for the socket (D4). **Partially superseded by §8 (2026-08-18):**
|
||||
the site→central stream hop now also runs over a real loopback HTTP/2 socket with
|
||||
the R2 batcher, protobuf serialization and Kestrel in the measured path. TLS and
|
||||
WAN latency remain out of frame.
|
||||
- **Not a real OPC UA run.** 375,000 monitored items were simulated at the adapter
|
||||
callback (D3). The DCL above that callback is real and fully exercised.
|
||||
- **Not 1 hour.** 20 minutes (D1). See finding F1 for the one question this leaves
|
||||
@@ -323,6 +326,111 @@ dotnet test tests/ZB.MOM.WW.ScadaBridge.PerformanceTests \
|
||||
`--help`-style knobs: `--sites`, `--instances-per-site`, `--tags-per-instance`,
|
||||
`--tag-update-period-seconds`, `--settle-minutes`, `--sustain-minutes`,
|
||||
`--sample-seconds`, `--health-interval-seconds`, `--debug-probe-interval-seconds`,
|
||||
`--subscribe-settle-seconds`, `--stream-probes-per-site`, `--sf-drain-messages`,
|
||||
`--slow-subscriber-events`, `--data-dir`, `--results`. An unknown key is rejected
|
||||
rather than ignored, so a typo cannot silently change the measured scale.
|
||||
`--subscribe-settle-seconds`, `--stream-probes-per-site`, `--wire-probes-per-site`,
|
||||
`--sf-drain-messages`, `--slow-subscriber-events`, `--data-dir`, `--results`. An
|
||||
unknown key is rejected rather than ignored, so a typo cannot silently change the
|
||||
measured scale.
|
||||
|
||||
Since 2026-08-18 the default protocol also attaches 10 **wire** stream subscribers
|
||||
per site (real Kestrel + real `SiteStreamGrpcClient`, §8); `--wire-probes-per-site 0`
|
||||
reproduces the original in-process-only shape.
|
||||
|
||||
---
|
||||
|
||||
## 8. Wire-path run — R2 stream batching in the hot path (2026-08-18)
|
||||
|
||||
**Raw metrics:** `docs/plans/2026-08-18-loadharness-results-wire.json`
|
||||
|
||||
### 8.1 Why this run exists
|
||||
|
||||
Both prior runs (§2 and the §4 F1 1-hour sustain) measured the stream path only up
|
||||
to the bounded send channel: `StreamSubscriberProbe` deliberately substitutes the
|
||||
final socket hop with a controllable reader (deviation D4 — required by the row-50
|
||||
slow-subscriber measurement). The R2 gRPC event batcher (`SiteStreamEventBatcher`,
|
||||
merged `c75c79f8` on 2026-08-15, about an hour *after* the §2 run) lives exactly in
|
||||
that substituted hop — so no benchmark run had ever exercised it, only R2's own
|
||||
component-level measurement (worst-case P99 25.4 ms) and trickle-rate live traffic
|
||||
on the rig (register row 48).
|
||||
|
||||
This run adds a **wire path** to the harness without touching the existing probes:
|
||||
each site hosts the production `SiteStreamGrpcServer` on a real Kestrel HTTP/2 (h2c)
|
||||
loopback listener (production `CommunicationOptions` defaults — batch cap 100,
|
||||
window 25 ms, channel capacity 1000), and the production `SiteStreamGrpcClient`
|
||||
dials in with `BatchingSupported = true`, one channel per site with 10 streams
|
||||
multiplexed on it — the exact production shape and the exact production code on both
|
||||
ends. Latency is recorded in the client callback, i.e. after protobuf
|
||||
deserialization, batch unpacking and domain mapping. The batch-size figures are the
|
||||
production `scadabridge.site.stream.batch_size` histogram itself, captured by a
|
||||
`MeterListener`. Deliberately not in the path: `ControlPlaneAuthInterceptor`
|
||||
(authenticates once per stream lifetime, nothing per-event) and TLS (production is
|
||||
h2c too).
|
||||
|
||||
The 100 wire subscribers are attached **in addition to** the 100 in-process probes,
|
||||
watching the same instances (same stride), so one run yields both distributions and
|
||||
their difference isolates the wire hop's cost. Total hub fan-out is therefore
|
||||
double that of the §2 run.
|
||||
|
||||
### 8.2 Environment
|
||||
|
||||
Identical posture to §1: same host (M4 Pro, 14 cores, 48 GB), .NET 10.0.5, server
|
||||
GC, Release; the 8-node `docker/` rig stayed running throughout (D8); 20-minute
|
||||
window after a 2-minute settle; harness at the state committed with this section.
|
||||
|
||||
### 8.3 Measured
|
||||
|
||||
| | in-process probes (§2 path) | **wire probes (new)** |
|
||||
|---|---|---|
|
||||
| Subscribers | 100 | 100 |
|
||||
| Events delivered (window) | 899,807 | **899,807** |
|
||||
| Events dropped / faults | 0 / — | **0 / 0** |
|
||||
| Latency mean | 1.79 ms | **28.00 ms** |
|
||||
| Latency P50 | 1.36 ms | **26.97 ms** |
|
||||
| Latency P95 | 4.19 ms | **32.07 ms** |
|
||||
| Latency P99 | 6.74 ms | **43.44 ms** |
|
||||
| Latency P99.9 / max | 25.83 / 32.17 ms | **53.94 / 62.42 ms** |
|
||||
|
||||
Batching (production histogram, window deltas):
|
||||
|
||||
```
|
||||
events 899,807 frames 12,462 → mean 72.20 events/frame (72x frame reduction)
|
||||
batch size p50/p95/p99/max = 75/75/75/75 batched frames 11,998 single-event frames 464
|
||||
server channel drops 0 stream faults 0 graceful completions 0
|
||||
```
|
||||
|
||||
Offered load held at **37,497/s = 100.0% of nominal** with the doubled fan-out
|
||||
(driver overrun 0.0 s, skipped 0). Whole-run CPU mean **66% of one core (4.7% of
|
||||
the box)** vs 41% in §2 — the added ~25% of one core buys 10 Kestrel listeners, 100
|
||||
live HTTP/2 streams, protobuf serialization of ~900 k events and the client-side
|
||||
unpack. Working-set slope **−116 MB/min** (a large mid-window release; peak
|
||||
3,413 MB), heap sawtooth with 5 gen-2 collects and endpoints 1,876→1,943 MB —
|
||||
consistent with the §4 F1 resolution (row 29: not a leak).
|
||||
|
||||
### 8.4 Reading the numbers
|
||||
|
||||
- **The batcher engaged exactly as designed, at its natural batch size.** The
|
||||
driver emits each instance's 75 tags contiguously within a slice, so a subscriber
|
||||
receives one 75-event burst per 10 s period. 75 < the 100-event cap, so the
|
||||
size cap never closes a batch early — nearly every burst becomes exactly one
|
||||
75-event frame (p50 = p99 = max = 75). The 464 single-event frames are burst
|
||||
stragglers that arrived after their batch's window closed.
|
||||
- **The ~26 ms wire P50 is dominated by the deliberate linger, and that is the
|
||||
documented trade.** Because bursts never hit the cap, every batch pays the full
|
||||
25 ms `GrpcStreamBatchWindow` — this workload is the batcher's *worst-case latency
|
||||
posture and best-case framing win* simultaneously. Wire P50 26.97 ms ≈ in-process
|
||||
P50 1.36 ms + the 25 ms window + sub-millisecond socket/serialize cost. Against
|
||||
the 250 ms P99 threshold the wire path passes with a **5.8× margin** (43.44 ms),
|
||||
and the `CommunicationOptionsValidator` bound (window strictly < 250 ms) is
|
||||
confirmed to compose: end-to-end latency stayed threshold-compliant with the
|
||||
window fully paid on every frame.
|
||||
- **Framing efficiency at scale is now measured, not extrapolated:** 899,807 events
|
||||
crossed the wire in 12,462 frames. Pre-R2, the same window would have cost
|
||||
899,807 frames.
|
||||
- **Nothing degraded elsewhere.** The in-process path under doubled fan-out reads
|
||||
P99 6.74 ms vs 4.57 ms in §2 — same order, still a 37× margin — and every §2
|
||||
pass criterion that this run re-exercises (offered load, zero drops at healthy
|
||||
subscribers, health reports, debug snapshots, S&F drain, slow-subscriber
|
||||
isolation) repeated its pass.
|
||||
|
||||
**Verdict: PASS.** R2's negotiated batching is now benchmark-proven at target scale
|
||||
over a real socket: 72× fewer frames, zero loss, zero faults, and worst-case-posture
|
||||
latency 5.8× inside the budget.
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,134 @@
|
||||
# High-Findings Remediation Plan — 2026-08-19 Arch Review
|
||||
|
||||
Scope: the four **High** findings from the 2026-08-19 updated architectural review
|
||||
(report artifact `43c7fa20-9626-4d59-a9c9-061524794660`; all four adversarially verified
|
||||
CONFIRMED). Baseline `main @ e9e44a9e`. Medium/Low findings and register-row reopens
|
||||
(95, 99) are explicitly out of scope for this plan.
|
||||
|
||||
Execution model: four file-disjoint tasks, one dedicated git worktree + subagent each,
|
||||
run in parallel; integration = cherry-pick onto main, combined build, affected test
|
||||
suites, `git diff` review, single merge session.
|
||||
|
||||
---
|
||||
|
||||
## T1 — P2-1 (High): Subscription completions carry no adapter generation
|
||||
|
||||
**File:** `src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Actors/DataConnectionActor.cs`
|
||||
(`SubscribeCompleted` ~:2806, `AlarmSubscribeCompleted` ~:2814 — actor-internal records,
|
||||
no wire-evolution constraint).
|
||||
|
||||
**Defect:** completions are applied unconditionally. An in-flight subscribe on
|
||||
generation N (window up to `SeedOverallTimeout` 120 s) racing a `SwapAdapter`
|
||||
(row-97 in-place reconfigure, failover) to generation N+1 causes:
|
||||
(a) stale completion stores a dead gen-N handle (~:1368), then the live chunk's
|
||||
duplicate guard (`_subscriptionIds.ContainsKey`, ~:2109) **releases the live monitored
|
||||
item** (~:2145) — tag reads resolved-forever, delivers nothing;
|
||||
(b) ~:1431 re-stamps old-endpoint seed values with current `_adapterGeneration`,
|
||||
bypassing the stale-data guard;
|
||||
(c) stale connection-level-failure rows knock a healthy connection back to Reconnecting (~:434);
|
||||
(d) alarm twin: a real completion releases the NEW live alarm feed as a duplicate (~:2548–2557).
|
||||
|
||||
**Fix:**
|
||||
- Add `AdapterGeneration` (stamped at subscribe launch) to both completion records;
|
||||
handlers drop mismatched completions entirely (no handle store, no seed apply, no
|
||||
failure-row processing, no duplicate-release).
|
||||
- On drop, release any successfully-created handles **against the captured adapter
|
||||
reference**, not the current one (also audit release sites ~:1295/:1404 — MxGateway
|
||||
int handles collide across sessions, so a handle must never be released against a
|
||||
different adapter instance than created it).
|
||||
- Do NOT widen scope to `ConnectResult` (P2-3, Med — separate follow-up).
|
||||
|
||||
**Tests (DataConnectionLayer.Tests):** stale `SubscribeCompleted` after a generation
|
||||
bump (1) does not overwrite/store handles, (2) does not release the live item,
|
||||
(3) does not seed stale values; alarm variant for `AlarmSubscribeCompleted`; released
|
||||
handles go to the captured adapter.
|
||||
|
||||
## T2 — P2-2 (High): MxGateway native alarm feed permanently dead after plain reconnect
|
||||
|
||||
**Files:** `src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Adapters/MxGatewayDataConnection.cs`
|
||||
(`ConnectAsync` :93–157; alarm state `_alarmCts`/`_alarmSubCount`/`_alarmSubSources`/
|
||||
`_alarmStreamPrefix`), `RealMxGatewayClient.cs` (:415–467 `RunAlarmStreamAsync` retry loop).
|
||||
|
||||
**Defect:** `ConnectAsync` resets the event loop/client/`_disconnectFired` but not the
|
||||
alarm-stream state (reset only in Disconnect/Dispose, which a reconnect never calls).
|
||||
After a gateway restart, `SubscribeAlarmsAsync` sees `_alarmCts != null && Covers` and
|
||||
returns success **without opening a stream on the new client**, while the orphaned
|
||||
`RunAlarmStreamAsync` retries the disposed old client every 5 s forever. The native-alarm
|
||||
mirror silently stays Uncertain until failover/reconfigure/redeploy.
|
||||
|
||||
**Fix:** in `ConnectAsync`, cancel + dispose `_alarmCts` and reset the alarm-subscription
|
||||
bookkeeping so the actor's post-reconnect re-subscribe opens a fresh stream on the new
|
||||
client; ensure the old stream task observes cancellation and exits (no zombie retry
|
||||
loop). Verify end-to-end that the actor's transparent re-subscribe path actually calls
|
||||
`SubscribeAlarmsAsync` again after reconnect and the feed resumes.
|
||||
|
||||
**Tests:** reconnect-then-alarm-resubscribe with a fake client: new stream opened on the
|
||||
new client, old loop terminated, alarm events flow again.
|
||||
|
||||
## T3 — P6-1 (High): Secured-write reject is not CAS-guarded
|
||||
|
||||
**Files:** `src/ZB.MOM.WW.ScadaBridge.ManagementService/ManagementActor.cs`
|
||||
(reject ~:1629–1655), `SecuredWriteRepository` (approve CAS `TryMarkApprovedAsync`
|
||||
:90–114 is the pattern; reject uses a blind full-row `UpdateAsync` :44–52), plus the
|
||||
repository interface.
|
||||
|
||||
**Defect:** handlers run concurrently (fire-and-forget PipeTo per envelope, :145–148),
|
||||
`PendingSecuredWrite` has no concurrency token. Reject reads Pending → approve's CAS
|
||||
wins and the multi-second `WriteTagAsync` relay runs → reject's blind update lands
|
||||
mid-relay, stamping Rejected while the plant write executes; approve's final blind
|
||||
update then overwrites back to Executed. The two-person-control record can be falsified
|
||||
in either direction.
|
||||
|
||||
**Fix:**
|
||||
- Add `TryMarkRejectedAsync` (`UPDATE … WHERE Status = 'Pending'`) mirroring the approve
|
||||
CAS; reject handler uses it and, on CAS failure, re-reads and returns a precise
|
||||
conflict error (same UX shape as approve's existing lost-race path).
|
||||
- Confirm approve's post-relay status write can no longer collide (once reject can only
|
||||
win from Pending, the relay leg owns the row) — if any other blind `UpdateAsync` on
|
||||
this row remains reachable concurrently, CAS it too.
|
||||
- Emit the `SecuredWriteReject` audit row only on CAS success.
|
||||
|
||||
**Tests (ManagementService.Tests):** reject-after-approve returns conflict and writes
|
||||
nothing; approve-after-reject returns conflict; repository-level CAS test (two
|
||||
transitions from one Pending row — exactly one wins).
|
||||
|
||||
## T4 — P6-2 (High): AlarmSummary page bypasses site scoping
|
||||
|
||||
**File:** `src/ZB.MOM.WW.ScadaBridge.CentralUI/Components/Pages/Monitoring/AlarmSummary.razor`
|
||||
(~:278). Pattern donors: `DebugView.razor` (:363/:427), `EventLogs.razor`, `ParkedMessages.razor`.
|
||||
|
||||
**Defect:** the page is `RequireDeployment`-gated (Deployer — exactly the role
|
||||
`SiteScopeService` scopes) but never injects `SiteScopeService`: `_sites =
|
||||
GetAllSitesAsync()` unfiltered, and no scope check guards site selection or
|
||||
`AlarmSummaryService.GetSiteAlarmsAsync`. A scoped Deployer can watch any site's live
|
||||
alarms including `CurrentValue`/`LimitValue` process data. Server-side gap (Blazor Server).
|
||||
|
||||
**Fix:** mirror the sibling pattern exactly — inject `SiteScopeService`, filter the site
|
||||
list, and enforce scope on the selected site before every alarm fetch (a crafted
|
||||
interaction naming an out-of-scope site must be denied server-side, not just hidden).
|
||||
|
||||
**Tests (CentralUI.Tests):** mirror whatever pins the siblings' scoping (site list
|
||||
filtered; out-of-scope fetch denied). If no sibling UI-scoping test exists, add a
|
||||
service/page-logic-level test rather than a new UI harness.
|
||||
|
||||
---
|
||||
|
||||
## Shared constraints (all tasks)
|
||||
|
||||
- Fix + tests only; no drive-by refactors. Match surrounding style/comment density.
|
||||
- Targeted verification per task: build the touched project(s) + test project(s), run
|
||||
the affected suites filtered — not the full solution sweep (that runs once at
|
||||
integration).
|
||||
- Each agent works ONLY inside its own worktree (absolute paths, `git -C`), commits with
|
||||
a pathspec limited to its own files, and reports the commit SHA + test results.
|
||||
- Update the relevant `docs/requirements/Component-*.md` only where it documents the now-
|
||||
contradicted behavior; one sentence, no restructuring.
|
||||
|
||||
## Integration
|
||||
|
||||
1. Cherry-pick the four commits onto `main` (file-disjoint; T1/T2 share a project but
|
||||
not files — verify no test-fixture collisions).
|
||||
2. `dotnet build ZB.MOM.WW.ScadaBridge.slnx`, then run DataConnectionLayer, Management-
|
||||
Service, and CentralUI test suites (plus any suites the agents touched).
|
||||
3. `git diff` review, commit plan doc + register/CLAUDE.md touch-ups if any, remove
|
||||
worktrees. No push without the user.
|
||||
@@ -168,7 +168,7 @@ The two-person authorization workflow for writes through the MxAccess Gateway. B
|
||||
|
||||
- **SubmitSecuredWriteCommand** (`Operator` role): validates the target connection exists and uses the **MxGateway** protocol (rejected otherwise), then inserts a `Pending` row capturing site / connection / tag / typed value / operator + comment. Nothing is written yet.
|
||||
- **ApproveSecuredWriteCommand** (`Verifier` role): enforces **no self-approval** (`VerifierUser ≠ OperatorUser`, checked **before** any state change) and a **compare-and-swap** race guard (`TryMarkApprovedAsync` — exactly one concurrent approver flips `Pending → Approved`; the loser is rejected). On winning, it validates the value type, decodes the value (guarded — a bad value fails the row deterministically rather than leaving it stuck `Approved`), and relays a `WriteTagRequest` to the site MxGateway via the Communication Layer, recording the outcome as `Executed` or `Failed`.
|
||||
- **RejectSecuredWriteCommand** (`Verifier` role): marks a `Pending` row `Rejected` with the verifier's reason.
|
||||
- **RejectSecuredWriteCommand** (`Verifier` role): marks a `Pending` row `Rejected` with the verifier's reason. Like approve, the transition is **compare-and-swap guarded** (`TryMarkRejectedAsync` — a conditional `UPDATE … WHERE Status = 'Pending'`, so approve, reject and the TTL expiry can never both transition one row); a loser records **nothing** — no status write and no `SecuredWriteReject` audit row — and is answered with the status that actually won. The blind full-row update it replaced could land after a concurrent approve had won its own CAS and while that approve's multi-second `WriteTagRequest` relay was still in flight, stamping the row `Rejected` for a write that physically executed at the plant (arch-review 2026-08-19 P6-1). Winning the approve CAS therefore gives that leg exclusive ownership of the row, which is what makes its post-relay `Executed`/`Failed` update safe as an unconditional write.
|
||||
- **ListSecuredWritesCommand** (any of `Operator` / `Verifier` / `Administrator`): pending queue + terminal history, global or per-site, **offset-paged** via `Skip` / `Take` (arch-review P3; additive — the defaults `Skip=0` / `Take=200` preserve the historical unpaged wire behaviour). The handler clamps `Take` to `1..500` and `Skip` to `>= 0`, calls `QueryAsync(Status, SiteId, skip, take)` for the page and `CountAsync(Status, SiteId)` for the unpaged total, and returns a `SecuredWriteListResult { Items, TotalCount }` so a client can render a pager. Although read-only, the history table exposes process-sensitive tag values, so it is **not** open to any authenticated user — it is gated to the two Secured Write roles plus `Administrator` (any-of), aligning with the `RequireSecuredWriteAccess` Central UI policy (arch-review UA1).
|
||||
|
||||
**Server-side pending TTL (arch-review S2).** A `Pending` secured write is not valid indefinitely. Both `ApproveSecuredWriteCommand` and `RejectSecuredWriteCommand` call a shared `TryExpireIfStaleAsync` helper immediately after the `Pending`-status check: if the row's `SubmittedAtUtc` is older than the configured `SecuredWritePendingTtl` (`ManagementServiceOptions`, default **24 h**), the helper atomically transitions it `Pending → Expired` via `ISecuredWriteRepository.TryMarkExpiredAsync` (compare-and-swap — no verifier recorded, `VerifierUser` stays `NULL`), emits a best-effort `SecuredWriteExpire` audit row (system actor, `Discarded` status — the closest terminal `AuditStatus`), and the command fails with an "expired" error. A stale setpoint therefore can **never** be approved or executed days later, and no `WriteTagRequest` is relayed. A non-positive TTL disables expiry. `ListSecuredWritesCommand` additionally runs an **opportunistic sweep**: on every read it walks the returned page and calls the same helper for each `Pending` row, so overdue writes surface as `Expired` the next time anyone views the queue — no dedicated expiry-timer singleton is needed because the ManagementActor already runs on every central node and the CAS makes concurrent multi-node sweeps idempotent.
|
||||
|
||||
@@ -11,6 +11,7 @@
|
||||
@inject IAlarmSummaryService AlarmSummaryService
|
||||
@inject ISiteRepository SiteRepository
|
||||
@inject ISiteAlarmLiveCache LiveAlarmCache
|
||||
@inject ZB.MOM.WW.ScadaBridge.CentralUI.Auth.SiteScopeService SiteScope
|
||||
|
||||
<div class="container-fluid mt-3" data-test="alarm-summary">
|
||||
<div class="d-flex justify-content-between align-items-center mb-3">
|
||||
@@ -275,7 +276,9 @@
|
||||
{
|
||||
try
|
||||
{
|
||||
_sites = await SiteRepository.GetAllSitesAsync();
|
||||
// Site scoping (CentralUI-002, arch-review P6-2): a scoped Deployment
|
||||
// user may only watch live alarms for the sites they are permitted on.
|
||||
_sites = await SiteScope.FilterSitesAsync(await SiteRepository.GetAllSitesAsync());
|
||||
}
|
||||
catch
|
||||
{
|
||||
@@ -283,17 +286,36 @@
|
||||
}
|
||||
}
|
||||
|
||||
// _sites is already filtered, so membership IS the scope check — mirrors
|
||||
// EventLogs.razor's SelectedSiteIsPermitted.
|
||||
private bool IsSiteSelectionPermitted(int siteId) => _sites.Any(s => s.Id == siteId);
|
||||
|
||||
private void ClearSiteSelection()
|
||||
{
|
||||
_selectedSiteId = null;
|
||||
_rows = Array.Empty<AlarmSummaryRow>();
|
||||
_notReporting = Array.Empty<string>();
|
||||
_rollup = new AlarmRollup(0, 0, 0, new Dictionary<AlarmKind, int>());
|
||||
StopTimer();
|
||||
DisposeLiveSubscription();
|
||||
}
|
||||
|
||||
private async Task OnSiteChangedAsync(ChangeEventArgs e)
|
||||
{
|
||||
var raw = e.Value?.ToString();
|
||||
if (string.IsNullOrEmpty(raw) || !int.TryParse(raw, out var siteId))
|
||||
{
|
||||
_selectedSiteId = null;
|
||||
_rows = Array.Empty<AlarmSummaryRow>();
|
||||
_notReporting = Array.Empty<string>();
|
||||
_rollup = new AlarmRollup(0, 0, 0, new Dictionary<AlarmKind, int>());
|
||||
StopTimer();
|
||||
DisposeLiveSubscription();
|
||||
ClearSiteSelection();
|
||||
return;
|
||||
}
|
||||
|
||||
// Site scoping (CentralUI-002, arch-review P6-2): the dropdown is already
|
||||
// filtered, but the selection must not be trusted on its own — re-check
|
||||
// server-side so a forged onchange value naming an out-of-scope site is
|
||||
// denied rather than merely hidden from the picker.
|
||||
if (!IsSiteSelectionPermitted(siteId))
|
||||
{
|
||||
ClearSiteSelection();
|
||||
return;
|
||||
}
|
||||
|
||||
@@ -319,6 +341,16 @@
|
||||
return;
|
||||
}
|
||||
|
||||
// Site scoping (CentralUI-002, arch-review P6-2): re-check before every
|
||||
// fetch, not just at selection time — defense in depth against
|
||||
// _selectedSiteId ever being set by a path that skips OnSiteChangedAsync's
|
||||
// check, so an out-of-scope site can never reach AlarmSummaryService.
|
||||
if (!IsSiteSelectionPermitted(siteId))
|
||||
{
|
||||
ClearSiteSelection();
|
||||
return;
|
||||
}
|
||||
|
||||
_loading = true;
|
||||
try
|
||||
{
|
||||
|
||||
@@ -86,6 +86,35 @@ public interface ISecuredWriteRepository
|
||||
DateTime decidedAtUtc,
|
||||
CancellationToken ct = default);
|
||||
|
||||
/// <summary>
|
||||
/// Atomically flips a row from <c>Pending</c> to <c>Rejected</c>, stamping the
|
||||
/// verifier identity, comment, and decision time, but ONLY if the row is still
|
||||
/// <c>Pending</c>. The reject counterpart of <see cref="TryMarkApprovedAsync"/>, and
|
||||
/// the compare-and-swap guard for the approve-vs-reject race: a stale-read reject
|
||||
/// used to land a blind full-row update while a concurrent approve's multi-second
|
||||
/// site relay was still in flight, marking the row <c>Rejected</c> while the plant
|
||||
/// write physically executed (arch-review 2026-08-19 P6-1). The conditional
|
||||
/// <c>WHERE Status='Pending'</c> makes exactly one of approve/reject/expiry win.
|
||||
/// The loser observes <c>false</c> and must record nothing — no status change and no
|
||||
/// reject audit row.
|
||||
/// </summary>
|
||||
/// <param name="id">Identity of the pending secured write.</param>
|
||||
/// <param name="verifierUser">The rejecting verifier's username.</param>
|
||||
/// <param name="verifierComment">Optional free-text comment from the verifier.</param>
|
||||
/// <param name="decidedAtUtc">UTC instant the rejection decision was made.</param>
|
||||
/// <param name="ct">Cancellation token.</param>
|
||||
/// <returns>
|
||||
/// A task resolving to <c>true</c> if this caller won the rejection (exactly one
|
||||
/// row transitioned), or <c>false</c> if the row was no longer <c>Pending</c>
|
||||
/// (already approved, rejected, or expired).
|
||||
/// </returns>
|
||||
Task<bool> TryMarkRejectedAsync(
|
||||
long id,
|
||||
string verifierUser,
|
||||
string? verifierComment,
|
||||
DateTime decidedAtUtc,
|
||||
CancellationToken ct = default);
|
||||
|
||||
/// <summary>
|
||||
/// Atomically flips a row from <c>Pending</c> to <c>Expired</c>, stamping the
|
||||
/// expiry as the decision time, but ONLY if the row is still <c>Pending</c>.
|
||||
|
||||
+29
@@ -113,6 +113,35 @@ WHERE Id = {id}
|
||||
return rowsAffected == 1;
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public async Task<bool> TryMarkRejectedAsync(
|
||||
long id,
|
||||
string verifierUser,
|
||||
string? verifierComment,
|
||||
DateTime decidedAtUtc,
|
||||
CancellationToken ct = default)
|
||||
{
|
||||
// Single-statement compare-and-swap, mirroring TryMarkApprovedAsync exactly.
|
||||
// A blind full-row UpdateAsync here used to be able to land AFTER a concurrent
|
||||
// approve had won its own CAS and while its multi-second WriteTagAsync relay was
|
||||
// still in flight — stamping the row Rejected even though the plant write was
|
||||
// physically executing, i.e. falsifying the two-person-control record
|
||||
// (arch-review 2026-08-19 P6-1). The conditional WHERE Status='Pending' makes
|
||||
// reject win only from Pending, so approve and reject can never both transition
|
||||
// one row. Only the columns the reject transition owns are written.
|
||||
var rowsAffected = await _context.Database.ExecuteSqlInterpolatedAsync(
|
||||
$@"UPDATE dbo.PendingSecuredWrites
|
||||
SET Status = 'Rejected',
|
||||
VerifierUser = {verifierUser},
|
||||
VerifierComment = {verifierComment},
|
||||
DecidedAtUtc = {decidedAtUtc}
|
||||
WHERE Id = {id}
|
||||
AND Status = 'Pending';",
|
||||
ct);
|
||||
|
||||
return rowsAffected == 1;
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public async Task<bool> TryMarkExpiredAsync(long id, DateTime expiredAtUtc, CancellationToken ct = default)
|
||||
{
|
||||
|
||||
@@ -983,7 +983,7 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers
|
||||
// dropped, and logs any tag that never yields a value.
|
||||
var seedValues = await SeedTagsAsync(adapter, tagsToSeed);
|
||||
|
||||
return new SubscribeCompleted(request, sender, results, seedValues);
|
||||
return new SubscribeCompleted(request, sender, results, seedValues, generation, adapter);
|
||||
}).PipeTo(self);
|
||||
}
|
||||
|
||||
@@ -1086,6 +1086,30 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers
|
||||
return Task.WhenAll(subscriptionIds.Select(id => adapter.UnsubscribeAsync(id)));
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Best-effort release of handles created by a subscribe whose adapter has since been
|
||||
/// swapped out. <paramref name="release"/> must target the adapter that CREATED the handles
|
||||
/// — they are adapter-scoped (MxGateway's are plain ints that collide across sessions), so
|
||||
/// releasing by handle against the replacement adapter can tear down a live subscription.
|
||||
/// The outgoing adapter is disposed fire-and-forget by <see cref="SwapAdapter"/> and may
|
||||
/// already be gone, so a failure here is expected and logged at Debug rather than surfaced.
|
||||
/// </summary>
|
||||
private void ReleaseStaleAgainstOwnAdapter(string description, Func<Task> release)
|
||||
{
|
||||
try
|
||||
{
|
||||
release().ContinueWith(
|
||||
t => _log.Debug("[{0}] Best-effort release of {1} failed: {2}",
|
||||
_connectionName, description, t.Exception?.GetBaseException().Message),
|
||||
TaskContinuationOptions.OnlyOnFaulted);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_log.Debug("[{0}] Best-effort release of {1} threw: {2}",
|
||||
_connectionName, description, ex.Message);
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Reads the current value of each tag so the Instance Actor
|
||||
/// has an initial value, retrying the still-empty subset a bounded number of times. A
|
||||
@@ -1256,6 +1280,51 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers
|
||||
private bool HandleSubscribeCompleted(SubscribeCompleted msg)
|
||||
{
|
||||
var instanceName = msg.Request.InstanceUniqueName;
|
||||
|
||||
// Results produced against an adapter that has since been swapped out (failover or
|
||||
// in-place reconfigure) are dropped WHOLE, mirroring the BatchSubscribeCompleted and
|
||||
// TagValueReceived generation guards. An initial subscribe can take up to
|
||||
// SeedOverallTimeout to complete, so this window is wide. Applying a stale completion
|
||||
// stores a dead session's handle under the tag's key, which then makes the LIVE
|
||||
// completion for the same tag look like a duplicate and get RELEASED (the tag reads
|
||||
// resolved forever and delivers nothing); re-stamps the old endpoint's seed values with
|
||||
// the CURRENT generation, so they bypass the stale-data guard and present as fresh; and
|
||||
// lets a connection-level failure row from the dead adapter knock the healthy new
|
||||
// connection back into Reconnecting.
|
||||
if (msg.AdapterGeneration != _adapterGeneration)
|
||||
{
|
||||
var staleIds = msg.Results
|
||||
.Where(r => r is { Success: true, AlreadySubscribed: false, SubscriptionId: not null })
|
||||
.Select(r => r.SubscriptionId!)
|
||||
.ToList();
|
||||
|
||||
_log.Debug(
|
||||
"[{0}] Dropping stale subscribe completion for instance {1} from adapter " +
|
||||
"generation {2} (current {3}); releasing {4} handle(s) against the adapter that created them",
|
||||
_connectionName, instanceName, msg.AdapterGeneration, _adapterGeneration, staleIds.Count);
|
||||
|
||||
// Release against msg.Adapter, NEVER _adapter: handles are adapter-scoped and
|
||||
// MxGateway's are plain ints that collide across sessions, so releasing by handle
|
||||
// against the replacement adapter can tear down a live subscription.
|
||||
ReleaseStaleAgainstOwnAdapter(
|
||||
$"{staleIds.Count} tag subscription handle(s) from adapter generation {msg.AdapterGeneration}",
|
||||
() => UnsubscribeIdsAsync(msg.Adapter, staleIds));
|
||||
|
||||
// _subscribesInFlight is deliberately NOT cleared here: ReSubscribeAll empties it
|
||||
// wholesale once the replacement adapter connects, and a marker the new generation
|
||||
// has since placed for the same tag must not be removed by this dead message.
|
||||
|
||||
// Reply failure so the requester re-sends. Nothing has registered this request's
|
||||
// tags in _subscriptionsByInstance (registration happens below, on the applied
|
||||
// path), so ReSubscribeAll cannot pick them up — the InstanceActor's per-connection
|
||||
// retry timer, armed until a Success arrives, is what re-establishes them.
|
||||
msg.ReplyTo.Tell(new SubscribeTagsResponse(
|
||||
msg.Request.CorrelationId, instanceName, false,
|
||||
"adapter replaced while the subscribe was in flight — will retry",
|
||||
DateTimeOffset.UtcNow));
|
||||
return false;
|
||||
}
|
||||
|
||||
if (!_subscriptionsByInstance.TryGetValue(instanceName, out var instanceTags))
|
||||
{
|
||||
// The instance was unsubscribed while the
|
||||
@@ -1292,7 +1361,10 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers
|
||||
}
|
||||
}
|
||||
|
||||
_ = UnsubscribeIdsAsync(_adapter, orphanedIds);
|
||||
// Against the adapter that created them (identical to _adapter here — the
|
||||
// generation guard above already proved the completion is current — but stating
|
||||
// the ownership rule at every release site is what keeps it true).
|
||||
_ = UnsubscribeIdsAsync(msg.Adapter, orphanedIds);
|
||||
|
||||
// The original sender is already gone (unsubscribed). Telling a dead
|
||||
// ref produces a dead letter, which is the harmless and observable
|
||||
@@ -1400,8 +1472,9 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers
|
||||
}
|
||||
}
|
||||
|
||||
// Fire-and-forget release of every handle this request created redundantly.
|
||||
_ = UnsubscribeIdsAsync(_adapter, redundantIds);
|
||||
// Fire-and-forget release of every handle this request created redundantly, against the
|
||||
// adapter that created them (see the orphaned-release site above).
|
||||
_ = UnsubscribeIdsAsync(msg.Adapter, redundantIds);
|
||||
|
||||
// Publish the resolution counts this request just changed. Without this push a
|
||||
// connection whose tags ALL resolve on the first attempt reported 0/0 forever:
|
||||
@@ -2502,14 +2575,46 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers
|
||||
alarmable.SubscribeAlarmsAsync(sourceRef, filter,
|
||||
t => self.Tell(new AlarmTransitionReceived(t, generation)))
|
||||
.ContinueWith(task => task.IsCompletedSuccessfully
|
||||
? new AlarmSubscribeCompleted(sourceRef, true, task.Result, null, subscriber, corr, inst) as object
|
||||
? new AlarmSubscribeCompleted(
|
||||
sourceRef, true, task.Result, null, subscriber, corr, inst, generation, alarmable) as object
|
||||
: new AlarmSubscribeCompleted(sourceRef, false, null,
|
||||
task.Exception?.GetBaseException().Message ?? "Unknown error", subscriber, corr, inst))
|
||||
task.Exception?.GetBaseException().Message ?? "Unknown error",
|
||||
subscriber, corr, inst, generation, alarmable))
|
||||
.PipeTo(self);
|
||||
}
|
||||
|
||||
private void HandleAlarmSubscribeCompleted(AlarmSubscribeCompleted msg)
|
||||
{
|
||||
// The alarm twin of HandleSubscribeCompleted's generation guard: a completion from an
|
||||
// adapter that has since been swapped out is dropped whole. Applying it would store a
|
||||
// dead feed's id — or, when the new adapter's feed is already stored, take the
|
||||
// duplicate branch and release the LIVE feed, leaving the source mirrored but silent.
|
||||
if (msg.AdapterGeneration != _adapterGeneration)
|
||||
{
|
||||
_log.Debug(
|
||||
"[{0}] Dropping stale alarm subscribe completion for source {1} from adapter " +
|
||||
"generation {2} (current {3})",
|
||||
_connectionName, msg.SourceReference, msg.AdapterGeneration, _adapterGeneration);
|
||||
|
||||
// Release the orphaned feed against the adapter that created it, never the current
|
||||
// one (see ReleaseStaleAgainstOwnAdapter).
|
||||
if (msg is { Success: true, SubscriptionId: not null })
|
||||
ReleaseStaleAgainstOwnAdapter(
|
||||
$"the alarm feed for {msg.SourceReference} from adapter generation {msg.AdapterGeneration}",
|
||||
() => msg.Adapter.UnsubscribeAlarmsAsync(msg.SubscriptionId));
|
||||
|
||||
// _alarmSubscribesInFlight is left untouched — ReSubscribeAllAlarms clears it
|
||||
// wholesale and may already have re-registered this source against the new adapter.
|
||||
// The subscriber state (_alarmSourceSubscribers) survives the swap, so the feed is
|
||||
// re-established by ReSubscribeAllAlarms; the failure reply just lets the
|
||||
// NativeAlarmActor retry rather than wait out its response timeout.
|
||||
msg.ReplyTo?.Tell(new SubscribeAlarmsResponse(
|
||||
msg.CorrelationId ?? string.Empty, msg.InstanceUniqueName ?? string.Empty,
|
||||
false, "adapter replaced while the alarm subscribe was in flight — will retry",
|
||||
DateTimeOffset.UtcNow));
|
||||
return;
|
||||
}
|
||||
|
||||
_alarmSubscribesInFlight.Remove(msg.SourceReference);
|
||||
|
||||
// The last (or only) subscriber may have been
|
||||
@@ -2521,15 +2626,16 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers
|
||||
// transitions to nobody for the lifetime of the adapter.
|
||||
if (!_alarmSourceSubscribers.ContainsKey(msg.SourceReference))
|
||||
{
|
||||
if (msg.Success && msg.SubscriptionId != null &&
|
||||
_adapter is IAlarmSubscribableConnection alarmable)
|
||||
if (msg.Success && msg.SubscriptionId != null)
|
||||
{
|
||||
_log.Warning(
|
||||
"[{0}] AlarmSubscribeCompleted arrived for source {1} but the last " +
|
||||
"subscriber unsubscribed while the subscribe was in flight; releasing " +
|
||||
"the orphaned adapter alarm feed.",
|
||||
_connectionName, msg.SourceReference);
|
||||
_ = alarmable.UnsubscribeAlarmsAsync(msg.SubscriptionId);
|
||||
// Against the adapter that created the feed (identical to _adapter here — the
|
||||
// generation guard above already proved the completion is current).
|
||||
_ = msg.Adapter.UnsubscribeAlarmsAsync(msg.SubscriptionId);
|
||||
}
|
||||
|
||||
// No live requester remains to receive a response.
|
||||
@@ -2547,14 +2653,11 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers
|
||||
// the stored id and leaking the already-tracked subscription.
|
||||
if (_alarmSubscriptionIds.ContainsKey(msg.SourceReference))
|
||||
{
|
||||
if (_adapter is IAlarmSubscribableConnection alarmable)
|
||||
{
|
||||
_log.Warning(
|
||||
"[{0}] Duplicate alarm feed for source {1}; releasing the redundant " +
|
||||
"subscription instead of overwriting the stored one.",
|
||||
_connectionName, msg.SourceReference);
|
||||
_ = alarmable.UnsubscribeAlarmsAsync(msg.SubscriptionId);
|
||||
}
|
||||
_log.Warning(
|
||||
"[{0}] Duplicate alarm feed for source {1}; releasing the redundant " +
|
||||
"subscription instead of overwriting the stored one.",
|
||||
_connectionName, msg.SourceReference);
|
||||
_ = msg.Adapter.UnsubscribeAlarmsAsync(msg.SubscriptionId);
|
||||
}
|
||||
else
|
||||
{
|
||||
@@ -2755,9 +2858,11 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers
|
||||
alarmable.SubscribeAlarmsAsync(sr, filter,
|
||||
t => self.Tell(new AlarmTransitionReceived(t, generation)))
|
||||
.ContinueWith(task => task.IsCompletedSuccessfully
|
||||
? new AlarmSubscribeCompleted(sr, true, task.Result, null, null, null, null) as object
|
||||
? new AlarmSubscribeCompleted(
|
||||
sr, true, task.Result, null, null, null, null, generation, alarmable) as object
|
||||
: new AlarmSubscribeCompleted(sr, false, null,
|
||||
task.Exception?.GetBaseException().Message ?? "Unknown error", null, null, null))
|
||||
task.Exception?.GetBaseException().Message ?? "Unknown error",
|
||||
null, null, null, generation, alarmable))
|
||||
.PipeTo(self);
|
||||
}
|
||||
}
|
||||
@@ -2803,16 +2908,30 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers
|
||||
internal record SubscribeTagResult(
|
||||
string TagPath, bool AlreadySubscribed, bool Success, string? SubscriptionId, string? Error,
|
||||
bool ConnectionLevelFailure = false);
|
||||
|
||||
/// <summary>
|
||||
/// Per-tag results of the initial subscribe issued off the actor thread by
|
||||
/// <see cref="HandleSubscribe"/>. Carries the adapter generation it ran against — and the
|
||||
/// adapter INSTANCE that produced the handles, because a stale completion's handles must be
|
||||
/// released against their own adapter (see <see cref="HandleSubscribeCompleted"/>).
|
||||
/// </summary>
|
||||
internal record SubscribeCompleted(
|
||||
SubscribeTagsRequest Request, IActorRef ReplyTo, IReadOnlyList<SubscribeTagResult> Results,
|
||||
IReadOnlyList<SeededValue> SeedValues);
|
||||
IReadOnlyList<SeededValue> SeedValues, int AdapterGeneration, IDataConnection Adapter);
|
||||
|
||||
/// <summary>An initial-read value captured during subscribe, delivered after the
|
||||
/// instance's tags are registered for fan-out.</summary>
|
||||
internal record SeededValue(string TagPath, TagValue Value);
|
||||
internal record AlarmTransitionReceived(NativeAlarmTransition Transition, int AdapterGeneration);
|
||||
|
||||
/// <summary>
|
||||
/// Result of an alarm-feed subscribe issued off the actor thread. Carries the adapter
|
||||
/// generation and the adapter instance for the same reason
|
||||
/// <see cref="SubscribeCompleted"/> does.
|
||||
/// </summary>
|
||||
internal record AlarmSubscribeCompleted(
|
||||
string SourceReference, bool Success, string? SubscriptionId, string? Error,
|
||||
IActorRef? ReplyTo, string? CorrelationId, string? InstanceUniqueName);
|
||||
IActorRef? ReplyTo, string? CorrelationId, string? InstanceUniqueName,
|
||||
int AdapterGeneration, IAlarmSubscribableConnection Adapter);
|
||||
public record GetHealthReport;
|
||||
}
|
||||
|
||||
@@ -133,6 +133,23 @@ public class MxGatewayDataConnection
|
||||
TaskContinuationOptions.OnlyOnFaulted);
|
||||
}
|
||||
|
||||
// Same reasoning as the event loop above, but for the native-alarm feed: without
|
||||
// this reset, the OLD RunAlarmStreamAsync loop keeps retrying against the now-
|
||||
// disposed previous client every 5s forever (it only exits on its own token), and
|
||||
// SubscribeAlarmsAsync's coverage check sees the stale _alarmCts/_alarmStreamPrefix
|
||||
// and short-circuits — so the transparent re-subscribe the actor issues after a
|
||||
// successful reconnect never opens a fresh stream on the NEW client. Mirror the
|
||||
// lock-guarded block already in DisconnectAsync/DisposeAsync.
|
||||
lock (_alarmLock)
|
||||
{
|
||||
_alarmCts?.Cancel();
|
||||
_alarmCts?.Dispose();
|
||||
_alarmCts = null;
|
||||
_alarmSubCount = 0;
|
||||
_alarmSubSources.Clear();
|
||||
_alarmStreamPrefix = string.Empty;
|
||||
}
|
||||
|
||||
Interlocked.Exchange(ref _disconnectFired, 0); // reset guard on (re)connect, like OPC UA
|
||||
_client = _clientFactory.Create();
|
||||
|
||||
|
||||
@@ -1604,6 +1604,14 @@ public class ManagementActor : ReceiveActor
|
||||
row.ExecutionError = error;
|
||||
|
||||
// UpdateAsync overwrites all columns -> pass the fully-populated entity.
|
||||
// Safe as a blind write (arch-review 2026-08-19 P6-1): winning the
|
||||
// Pending->Approved CAS above gives this leg EXCLUSIVE ownership of the row from
|
||||
// that point on. Every other writer transitions only out of Pending —
|
||||
// TryMarkApprovedAsync, TryMarkRejectedAsync and TryMarkExpiredAsync (both the
|
||||
// approve/reject inline expiry and the opportunistic list sweep) are all
|
||||
// conditional on Status='Pending' — so nothing can overwrite the row while the
|
||||
// multi-second relay above is in flight. Do not reintroduce an unconditional
|
||||
// transition on this table.
|
||||
await repo.UpdateAsync(row);
|
||||
|
||||
// Terminal execute outcome: Delivered (relay succeeded) maps to canonical
|
||||
@@ -1646,15 +1654,37 @@ public class ManagementActor : ReceiveActor
|
||||
throw new ManagementCommandException(
|
||||
"A secured write cannot be verified by its submitter.");
|
||||
|
||||
// Compare-and-swap: guards the approve-vs-reject race exactly as approve's own
|
||||
// CAS guards the two-verifier race (arch-review 2026-08-19 P6-1). The status
|
||||
// check above is a stale read — commands run concurrently, each on its own DI
|
||||
// scope and across both central nodes on one shared MS SQL — so a concurrent
|
||||
// approve can win its CAS and enter its multi-second WriteTagAsync relay in the
|
||||
// gap. The blind full-row UpdateAsync that used to sit here would then land
|
||||
// mid-relay and stamp the row Rejected while the plant write physically
|
||||
// executed, falsifying the two-person-control record. Reject now transitions
|
||||
// ONLY from Pending; the loser records nothing.
|
||||
var decidedAtUtc = DateTime.UtcNow;
|
||||
if (!await repo.TryMarkRejectedAsync(cmd.Id, user.Username, cmd.Comment, decidedAtUtc))
|
||||
{
|
||||
// Re-read so the conflict names the status that actually won (approve,
|
||||
// another reject, or the TTL expiry) — same UX shape as the stale-read
|
||||
// guard above and approve's own lost-race error.
|
||||
var current = await repo.GetAsync(cmd.Id);
|
||||
throw new ManagementCommandException(current is null
|
||||
? $"Secured write {cmd.Id} is no longer pending — already decided."
|
||||
: $"Secured write {cmd.Id} is '{current.Status}', not Pending; it cannot be rejected.");
|
||||
}
|
||||
|
||||
// We won the race. Stamp the decision locally so the returned DTO carries the
|
||||
// same values the CAS committed.
|
||||
entity.Status = "Rejected";
|
||||
entity.VerifierUser = user.Username;
|
||||
entity.VerifierComment = cmd.Comment;
|
||||
entity.DecidedAtUtc = DateTime.UtcNow;
|
||||
entity.DecidedAtUtc = decidedAtUtc;
|
||||
|
||||
// UpdateAsync overwrites all columns -> pass the fully-populated entity.
|
||||
await repo.UpdateAsync(entity);
|
||||
|
||||
// Reject is a terminal lifecycle event (canonical Discarded outcome).
|
||||
// Reject is a terminal lifecycle event (canonical Discarded outcome). Emitted
|
||||
// ONLY on a won CAS, so a lost race never leaves a reject row in the audit log
|
||||
// for a write that was actually approved and executed.
|
||||
// Actor = the verifier; the operator is carried in the row's extra detail.
|
||||
await EmitSecuredWriteAuditAsync(
|
||||
sp, AuditKind.SecuredWriteReject, AuditStatus.Discarded, entity, actor: user.Username);
|
||||
|
||||
@@ -67,6 +67,7 @@ public class AlarmSummaryRenderTests : BunitContext
|
||||
Services.AddSingleton(_siteRepo);
|
||||
|
||||
Services.AddSingleton<ISiteAlarmLiveCache>(_liveCache);
|
||||
Services.AddScoped<ZB.MOM.WW.ScadaBridge.CentralUI.Auth.SiteScopeService>();
|
||||
|
||||
var claims = new[]
|
||||
{
|
||||
@@ -278,6 +279,53 @@ public class AlarmSummaryRenderTests : BunitContext
|
||||
});
|
||||
}
|
||||
|
||||
// ── arch-review P6-2: site scoping (a scoped Deployer must not watch another
|
||||
// site's live alarms — the page was missed by the ad0100f2 scoping survey) ──
|
||||
|
||||
[Fact]
|
||||
public void SiteScoping_ScopedDeploymentUser_SeesOnlyPermittedSiteInPicker()
|
||||
{
|
||||
var scopedUser = new ClaimsPrincipal(new ClaimsIdentity(new[]
|
||||
{
|
||||
new Claim(JwtTokenService.UsernameClaimType, "scoped"),
|
||||
new Claim(JwtTokenService.RoleClaimType, "Deployer"),
|
||||
new Claim(JwtTokenService.SiteIdClaimType, "1"), // Site 1 only
|
||||
}, "TestAuth"));
|
||||
Services.AddSingleton<AuthenticationStateProvider>(new TestAuthStateProvider(scopedUser));
|
||||
|
||||
var cut = Render<AlarmSummaryPage>();
|
||||
|
||||
var options = cut.FindAll("[data-test='alarm-summary-site'] option");
|
||||
// "Select site…" placeholder + Site 1 only — Site 2 must not appear.
|
||||
Assert.Equal(2, options.Count);
|
||||
Assert.DoesNotContain("Site 2", cut.Markup);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void SiteScoping_ForgedOutOfScopeSiteId_IsDeniedNotFetched()
|
||||
{
|
||||
// The picker never renders Site 2 for this user, but the change handler
|
||||
// must independently reject an out-of-scope id arriving via any onchange
|
||||
// value (e.g. a forged one, since Blazor Server processes the SignalR
|
||||
// event server-side) rather than trusting it because it came through the
|
||||
// <select> element.
|
||||
var scopedUser = new ClaimsPrincipal(new ClaimsIdentity(new[]
|
||||
{
|
||||
new Claim(JwtTokenService.UsernameClaimType, "scoped"),
|
||||
new Claim(JwtTokenService.RoleClaimType, "Deployer"),
|
||||
new Claim(JwtTokenService.SiteIdClaimType, "1"), // Site 1 only
|
||||
}, "TestAuth"));
|
||||
Services.AddSingleton<AuthenticationStateProvider>(new TestAuthStateProvider(scopedUser));
|
||||
|
||||
var cut = Render<AlarmSummaryPage>();
|
||||
cut.Find("[data-test='alarm-summary-site']").Change("2"); // out-of-scope, forged
|
||||
|
||||
// Denied server-side: no alarm table renders, selection resets to empty.
|
||||
Assert.Empty(cut.FindAll("tr[data-test='alarm-summary-row']"));
|
||||
Assert.NotNull(cut.Find("[data-test='alarm-summary-empty']"));
|
||||
_summary.DidNotReceive().GetSiteAlarmsAsync(2, Arg.Any<CancellationToken>());
|
||||
}
|
||||
|
||||
// ── arch-review R2 N7: disposal guard on the live callback ─────────────────
|
||||
|
||||
[Fact]
|
||||
|
||||
@@ -49,6 +49,7 @@ public class AlarmSummaryVirtualizeTests : BunitContext
|
||||
}));
|
||||
Services.AddSingleton(_siteRepo);
|
||||
Services.AddSingleton<ISiteAlarmLiveCache>(new InertLiveCache());
|
||||
Services.AddScoped<ZB.MOM.WW.ScadaBridge.CentralUI.Auth.SiteScopeService>();
|
||||
|
||||
var claims = new[]
|
||||
{
|
||||
|
||||
+102
@@ -148,6 +148,108 @@ public class SecuredWriteRepositoryTests : IClassFixture<MsSqlMigrationFixture>
|
||||
Assert.False(await repo.TryMarkExpiredAsync(id, DateTime.UtcNow));
|
||||
}
|
||||
|
||||
[SkippableFact]
|
||||
public async Task TryMarkRejectedAsync_PendingRow_FlipsToRejected_Once()
|
||||
{
|
||||
Skip.IfNot(_fixture.Available, _fixture.SkipReason);
|
||||
|
||||
var siteId = NewSiteId();
|
||||
await using var context = CreateContext();
|
||||
var repo = new SecuredWriteRepository(context);
|
||||
|
||||
var id = await repo.AddAsync(NewRow(siteId, status: "Pending"));
|
||||
|
||||
Assert.True(await repo.TryMarkRejectedAsync(id, "ver.bob", "not authorized", DateTime.UtcNow));
|
||||
// CAS: a second caller loses because the row is no longer Pending.
|
||||
Assert.False(await repo.TryMarkRejectedAsync(id, "ver.carol", "also no", DateTime.UtcNow));
|
||||
|
||||
await using var readContext = CreateContext();
|
||||
var row = await new SecuredWriteRepository(readContext).GetAsync(id);
|
||||
Assert.NotNull(row);
|
||||
Assert.Equal("Rejected", row!.Status);
|
||||
Assert.Equal("ver.bob", row.VerifierUser);
|
||||
Assert.Equal("not authorized", row.VerifierComment);
|
||||
Assert.NotNull(row.DecidedAtUtc);
|
||||
// Reject writes only the columns its transition owns — the operator's submission
|
||||
// and the (never-run) execution columns are untouched.
|
||||
Assert.Equal("op.alice", row.OperatorUser);
|
||||
Assert.Null(row.ExecutedAtUtc);
|
||||
Assert.Null(row.ExecutionError);
|
||||
}
|
||||
|
||||
[SkippableFact]
|
||||
public async Task ApproveThenReject_OnOnePendingRow_ApproveWins_RejectReturnsFalse()
|
||||
{
|
||||
// arch-review 2026-08-19 P6-1: approve and reject both transition out of Pending,
|
||||
// so exactly one may win. This is the ordering that used to falsify the
|
||||
// two-person-control record — a blind reject update landing after approve had
|
||||
// already committed (and, in production, while its multi-second site relay was
|
||||
// still in flight) rewrote the row to Rejected for a write that executed.
|
||||
Skip.IfNot(_fixture.Available, _fixture.SkipReason);
|
||||
|
||||
var siteId = NewSiteId();
|
||||
await using var context = CreateContext();
|
||||
var repo = new SecuredWriteRepository(context);
|
||||
|
||||
var id = await repo.AddAsync(NewRow(siteId, status: "Pending"));
|
||||
|
||||
Assert.True(await repo.TryMarkApprovedAsync(id, "ver.bob", "ok", DateTime.UtcNow));
|
||||
Assert.False(await repo.TryMarkRejectedAsync(id, "ver.carol", "no", DateTime.UtcNow));
|
||||
|
||||
await using var readContext = CreateContext();
|
||||
var row = await new SecuredWriteRepository(readContext).GetAsync(id);
|
||||
Assert.NotNull(row);
|
||||
// The row reflects the winner ONLY — no trace of the losing reject.
|
||||
Assert.Equal("Approved", row!.Status);
|
||||
Assert.Equal("ver.bob", row.VerifierUser);
|
||||
Assert.Equal("ok", row.VerifierComment);
|
||||
}
|
||||
|
||||
[SkippableFact]
|
||||
public async Task RejectThenApprove_OnOnePendingRow_RejectWins_ApproveReturnsFalse()
|
||||
{
|
||||
// The opposite order: once rejected, the row can never be approved (and therefore
|
||||
// never relayed to the plant).
|
||||
Skip.IfNot(_fixture.Available, _fixture.SkipReason);
|
||||
|
||||
var siteId = NewSiteId();
|
||||
await using var context = CreateContext();
|
||||
var repo = new SecuredWriteRepository(context);
|
||||
|
||||
var id = await repo.AddAsync(NewRow(siteId, status: "Pending"));
|
||||
|
||||
Assert.True(await repo.TryMarkRejectedAsync(id, "ver.carol", "no", DateTime.UtcNow));
|
||||
Assert.False(await repo.TryMarkApprovedAsync(id, "ver.bob", "ok", DateTime.UtcNow));
|
||||
|
||||
await using var readContext = CreateContext();
|
||||
var row = await new SecuredWriteRepository(readContext).GetAsync(id);
|
||||
Assert.NotNull(row);
|
||||
Assert.Equal("Rejected", row!.Status);
|
||||
Assert.Equal("ver.carol", row.VerifierUser);
|
||||
Assert.Equal("no", row.VerifierComment);
|
||||
}
|
||||
|
||||
[SkippableFact]
|
||||
public async Task TryMarkRejectedAsync_ExpiredRow_ReturnsFalse()
|
||||
{
|
||||
Skip.IfNot(_fixture.Available, _fixture.SkipReason);
|
||||
|
||||
var siteId = NewSiteId();
|
||||
await using var context = CreateContext();
|
||||
var repo = new SecuredWriteRepository(context);
|
||||
|
||||
var id = await repo.AddAsync(NewRow(siteId, status: "Pending"));
|
||||
Assert.True(await repo.TryMarkExpiredAsync(id, DateTime.UtcNow));
|
||||
|
||||
// Already Expired -> the conditional WHERE Status='Pending' matches no row.
|
||||
Assert.False(await repo.TryMarkRejectedAsync(id, "ver.bob", null, DateTime.UtcNow));
|
||||
|
||||
await using var readContext = CreateContext();
|
||||
var row = await new SecuredWriteRepository(readContext).GetAsync(id);
|
||||
Assert.Equal("Expired", row!.Status);
|
||||
Assert.Null(row.VerifierUser);
|
||||
}
|
||||
|
||||
[SkippableFact]
|
||||
public async Task CountAsync_FiltersLikeQueryAsync()
|
||||
{
|
||||
|
||||
+360
@@ -0,0 +1,360 @@
|
||||
using Akka.Actor;
|
||||
using Akka.TestKit.Xunit2;
|
||||
using NSubstitute;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Protocol;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Messages.DataConnection;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Types.Alarms;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums;
|
||||
using ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Actors;
|
||||
using ZB.MOM.WW.ScadaBridge.HealthMonitoring;
|
||||
|
||||
namespace ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests.Actors;
|
||||
|
||||
/// <summary>
|
||||
/// Arch-review 2026-08-19 <b>P2-1</b> — an in-flight subscribe that completes AFTER the adapter
|
||||
/// has been swapped out (failover, or the in-place reconfigure of register row 97).
|
||||
///
|
||||
/// <para>
|
||||
/// <b>The gap these pin.</b> <c>SubscribeCompleted</c>/<c>AlarmSubscribeCompleted</c> carried no
|
||||
/// adapter generation, and their handlers applied them unconditionally — while an initial
|
||||
/// subscribe can stay in flight for up to <c>SeedOverallTimeout</c> (120 s), so the window is
|
||||
/// wide. A stale completion stored a DEAD session's handle under the tag's key, which then made
|
||||
/// the live completion for the same tag look like a duplicate and get released: the tag reads
|
||||
/// resolved forever and delivers nothing. It also re-stamped the old endpoint's seed values with
|
||||
/// the CURRENT generation (bypassing the stale-data guard) and let a connection-level failure row
|
||||
/// from the dead adapter knock the healthy new connection back into Reconnecting. The alarm twin
|
||||
/// released the live alarm feed the same way.
|
||||
/// </para>
|
||||
///
|
||||
/// <para>
|
||||
/// Handles are adapter-scoped — MxGateway's are plain ints that collide across sessions — so a
|
||||
/// dropped completion's handles are released against the adapter that CREATED them, never the
|
||||
/// current one.
|
||||
/// </para>
|
||||
/// </summary>
|
||||
[Collection("DataConnectionManagerActor")]
|
||||
public class DataConnectionStaleCompletionTests : TestKit
|
||||
{
|
||||
private const string OldEndpoint = "opc.tcp://old-plc:4840";
|
||||
private const string NewEndpoint = "opc.tcp://new-plc:4840";
|
||||
private const string Tag = "sensor/temp";
|
||||
private const string Source = "Tank01";
|
||||
|
||||
private readonly ISiteHealthCollector _health = Substitute.For<ISiteHealthCollector>();
|
||||
private readonly IDataConnectionFactory _factory = Substitute.For<IDataConnectionFactory>();
|
||||
|
||||
private readonly DataConnectionOptions _options = new()
|
||||
{
|
||||
ReconnectInterval = TimeSpan.FromMilliseconds(100),
|
||||
TagResolutionRetryInterval = TimeSpan.FromMilliseconds(200),
|
||||
WriteTimeout = TimeSpan.FromSeconds(5),
|
||||
SeedReadMaxAttempts = 1,
|
||||
SeedReadRetryDelay = TimeSpan.FromMilliseconds(10)
|
||||
};
|
||||
|
||||
private static Dictionary<string, string> Config(string endpoint) =>
|
||||
new() { ["Endpoint"] = endpoint };
|
||||
|
||||
private IActorRef CreateConnectionActor(string name, IDataConnection adapter) =>
|
||||
Sys.ActorOf(Props.Create(() =>
|
||||
new DataConnectionActor(
|
||||
name, adapter, _options, _health, _factory, "OpcUa",
|
||||
Config(OldEndpoint), null, 3)), name);
|
||||
|
||||
private void ExpectNewAdapter(IDataConnection adapter) =>
|
||||
_factory.Create("OpcUa", Arg.Is<IDictionary<string, string>>(d => d["Endpoint"] == NewEndpoint))
|
||||
.Returns(adapter);
|
||||
|
||||
private static IDataConnection HealthyAdapter(string subscriptionId)
|
||||
{
|
||||
var adapter = Substitute.For<IDataConnection>();
|
||||
adapter.ConnectAsync(Arg.Any<IDictionary<string, string>>(), Arg.Any<CancellationToken>())
|
||||
.Returns(Task.CompletedTask);
|
||||
adapter.Status.Returns(ConnectionHealth.Connected);
|
||||
adapter.SubscribeAsync(Arg.Any<string>(), Arg.Any<SubscriptionCallback>(), Arg.Any<CancellationToken>())
|
||||
.Returns(subscriptionId);
|
||||
adapter.ReadAsync(Arg.Any<string>(), Arg.Any<CancellationToken>())
|
||||
.Returns(new ReadResult(false, null, null));
|
||||
return adapter;
|
||||
}
|
||||
|
||||
private static IDataConnection HealthyAlarmAdapter(string alarmSubscriptionId)
|
||||
{
|
||||
var adapter = Substitute.For<IDataConnection, IAlarmSubscribableConnection>();
|
||||
adapter.ConnectAsync(Arg.Any<IDictionary<string, string>>(), Arg.Any<CancellationToken>())
|
||||
.Returns(Task.CompletedTask);
|
||||
adapter.Status.Returns(ConnectionHealth.Connected);
|
||||
((IAlarmSubscribableConnection)adapter)
|
||||
.SubscribeAlarmsAsync(Arg.Any<string>(), Arg.Any<string?>(),
|
||||
Arg.Any<AlarmTransitionCallback>(), Arg.Any<CancellationToken>())
|
||||
.Returns(Task.FromResult(alarmSubscriptionId));
|
||||
return adapter;
|
||||
}
|
||||
|
||||
private static NativeAlarmTransition Raise(string sourceRef, string sourceObj) =>
|
||||
new(sourceRef, sourceObj, "AnalogLimit.Hi", AlarmTransitionKind.Raise,
|
||||
new AlarmConditionState(true, false, null, AlarmShelveState.Unshelved, false, 500),
|
||||
"Process", "hi", "hi", "", "", null, DateTimeOffset.UtcNow, "92", "90");
|
||||
|
||||
private int HealthTransitions(string connectionName, ConnectionHealth health) =>
|
||||
_health.ReceivedCalls().Count(c =>
|
||||
c.GetMethodInfo().Name == nameof(ISiteHealthCollector.UpdateConnectionHealth) &&
|
||||
c.GetArguments() is [string name, ConnectionHealth reported, ..] &&
|
||||
name == connectionName && reported == health);
|
||||
|
||||
private static bool CalledWith(IDataConnection adapter, string method, string firstArg) =>
|
||||
adapter.ReceivedCalls().Any(c =>
|
||||
c.GetMethodInfo().Name == method && c.GetArguments()[0] as string == firstArg);
|
||||
|
||||
private static bool Called(IDataConnection adapter, string method) =>
|
||||
adapter.ReceivedCalls().Any(c => c.GetMethodInfo().Name == method);
|
||||
|
||||
// ── Tag subscribe ──
|
||||
|
||||
/// <summary>
|
||||
/// The headline data-loss case: the stale completion must not park its dead handle under the
|
||||
/// tag's key, because the live completion for the same tag then reads as a duplicate and has
|
||||
/// its (working) monitored item released — leaving the tag permanently resolved and silent.
|
||||
///
|
||||
/// <para>
|
||||
/// <b>Fails before the fix:</b> the stale completion stored <c>sub-old</c>, and the live
|
||||
/// handle <c>sub-new</c> was released against the new adapter.
|
||||
/// </para>
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public void StaleSubscribeCompleted_DoesNotStoreItsHandle_AndLeavesTheLiveMonitoredItemAlone()
|
||||
{
|
||||
const string name = "stale-sub-handle";
|
||||
var gate = new TaskCompletionSource<string>();
|
||||
|
||||
var oldAdapter = HealthyAdapter("sub-old");
|
||||
oldAdapter.SubscribeAsync(Tag, Arg.Any<SubscriptionCallback>(), Arg.Any<CancellationToken>())
|
||||
.Returns(_ => gate.Task);
|
||||
var newAdapter = HealthyAdapter("sub-new");
|
||||
ExpectNewAdapter(newAdapter);
|
||||
|
||||
var actor = CreateConnectionActor(name, oldAdapter);
|
||||
AwaitCondition(() => HealthTransitions(name, ConnectionHealth.Connected) >= 1,
|
||||
TimeSpan.FromSeconds(5), TimeSpan.FromMilliseconds(10));
|
||||
|
||||
var probe = CreateTestProbe();
|
||||
probe.Send(actor, new SubscribeTagsRequest("c1", "inst1", name, [Tag], DateTimeOffset.UtcNow));
|
||||
AwaitCondition(() => CalledWith(oldAdapter, nameof(IDataConnection.SubscribeAsync), Tag),
|
||||
TimeSpan.FromSeconds(5));
|
||||
|
||||
actor.Tell(new ReconfigureConnectionCommand(name, "OpcUa", Config(NewEndpoint), null, 3));
|
||||
AwaitCondition(() => HealthTransitions(name, ConnectionHealth.Connected) >= 2,
|
||||
TimeSpan.FromSeconds(5), TimeSpan.FromMilliseconds(10));
|
||||
|
||||
// The Instance Actor's retry — the live subscribe, against the new adapter.
|
||||
probe.Send(actor, new SubscribeTagsRequest("c2", "inst1", name, [Tag], DateTimeOffset.UtcNow));
|
||||
probe.FishForMessage<SubscribeTagsResponse>(
|
||||
m => m.CorrelationId == "c2" && m.Success, TimeSpan.FromSeconds(5));
|
||||
|
||||
// Only now does the generation-N subscribe answer.
|
||||
gate.SetResult("sub-old");
|
||||
AwaitCondition(() => CalledWith(oldAdapter, nameof(IDataConnection.UnsubscribeAsync), "sub-old"),
|
||||
TimeSpan.FromSeconds(5), TimeSpan.FromMilliseconds(10));
|
||||
|
||||
Assert.False(Called(newAdapter, nameof(IDataConnection.UnsubscribeAsync)));
|
||||
|
||||
// The registry still holds the LIVE handle: the teardown releases sub-new, not sub-old.
|
||||
probe.Send(actor, new UnsubscribeTagsRequest("c3", "inst1", name, DateTimeOffset.UtcNow));
|
||||
AwaitCondition(() => CalledWith(newAdapter, nameof(IDataConnection.UnsubscribeAsync), "sub-new"),
|
||||
TimeSpan.FromSeconds(5), TimeSpan.FromMilliseconds(10));
|
||||
Assert.False(CalledWith(newAdapter, nameof(IDataConnection.UnsubscribeAsync), "sub-old"));
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// The handles a dropped completion did create are released — against the adapter that
|
||||
/// created them. Releasing them against the CURRENT adapter is not a harmless mismatch:
|
||||
/// MxGateway subscription handles are plain ints reused across sessions, so the release
|
||||
/// would land on whatever the new session happens to have under that number.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public void StaleSubscribeCompleted_ReleasesItsHandles_AgainstTheAdapterThatCreatedThem()
|
||||
{
|
||||
const string name = "stale-sub-release";
|
||||
var gate = new TaskCompletionSource<string>();
|
||||
|
||||
var oldAdapter = HealthyAdapter("sub-old");
|
||||
oldAdapter.SubscribeAsync(Tag, Arg.Any<SubscriptionCallback>(), Arg.Any<CancellationToken>())
|
||||
.Returns(_ => gate.Task);
|
||||
var newAdapter = HealthyAdapter("sub-new");
|
||||
ExpectNewAdapter(newAdapter);
|
||||
|
||||
var actor = CreateConnectionActor(name, oldAdapter);
|
||||
AwaitCondition(() => HealthTransitions(name, ConnectionHealth.Connected) >= 1,
|
||||
TimeSpan.FromSeconds(5), TimeSpan.FromMilliseconds(10));
|
||||
|
||||
var probe = CreateTestProbe();
|
||||
probe.Send(actor, new SubscribeTagsRequest("c1", "inst1", name, [Tag], DateTimeOffset.UtcNow));
|
||||
AwaitCondition(() => CalledWith(oldAdapter, nameof(IDataConnection.SubscribeAsync), Tag),
|
||||
TimeSpan.FromSeconds(5));
|
||||
|
||||
actor.Tell(new ReconfigureConnectionCommand(name, "OpcUa", Config(NewEndpoint), null, 3));
|
||||
AwaitCondition(() => HealthTransitions(name, ConnectionHealth.Connected) >= 2,
|
||||
TimeSpan.FromSeconds(5), TimeSpan.FromMilliseconds(10));
|
||||
|
||||
gate.SetResult("sub-old");
|
||||
|
||||
AwaitCondition(() => CalledWith(oldAdapter, nameof(IDataConnection.UnsubscribeAsync), "sub-old"),
|
||||
TimeSpan.FromSeconds(5), TimeSpan.FromMilliseconds(10));
|
||||
Assert.False(Called(newAdapter, nameof(IDataConnection.UnsubscribeAsync)));
|
||||
|
||||
// The requester is told the subscribe did not take, so its retry timer re-sends: the
|
||||
// dropped completion registered nothing, so the reconnect re-subscribe cannot cover it.
|
||||
probe.FishForMessage<SubscribeTagsResponse>(
|
||||
m => m.CorrelationId == "c1" && !m.Success, TimeSpan.FromSeconds(5));
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// The seed values a dropped completion carries are readings from the endpoint the operator
|
||||
/// has just corrected AWAY from. Applying them re-stamped each one with the CURRENT adapter
|
||||
/// generation, which is precisely what the generation guard on <c>TagValueReceived</c>
|
||||
/// exists to stop — the old endpoint's process values presented as fresh.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public void StaleSubscribeCompleted_DoesNotDeliverItsSeedValues()
|
||||
{
|
||||
const string name = "stale-sub-seed";
|
||||
var gate = new TaskCompletionSource<string>();
|
||||
|
||||
var oldAdapter = HealthyAdapter("sub-old");
|
||||
oldAdapter.SubscribeAsync(Tag, Arg.Any<SubscriptionCallback>(), Arg.Any<CancellationToken>())
|
||||
.Returns(_ => gate.Task);
|
||||
oldAdapter.ReadAsync(Tag, Arg.Any<CancellationToken>())
|
||||
.Returns(new ReadResult(true, new TagValue(999, QualityCode.Good, DateTimeOffset.UtcNow), null));
|
||||
var newAdapter = HealthyAdapter("sub-new");
|
||||
ExpectNewAdapter(newAdapter);
|
||||
|
||||
var actor = CreateConnectionActor(name, oldAdapter);
|
||||
AwaitCondition(() => HealthTransitions(name, ConnectionHealth.Connected) >= 1,
|
||||
TimeSpan.FromSeconds(5), TimeSpan.FromMilliseconds(10));
|
||||
|
||||
var probe = CreateTestProbe();
|
||||
probe.Send(actor, new SubscribeTagsRequest("c1", "inst1", name, [Tag], DateTimeOffset.UtcNow));
|
||||
AwaitCondition(() => CalledWith(oldAdapter, nameof(IDataConnection.SubscribeAsync), Tag),
|
||||
TimeSpan.FromSeconds(5));
|
||||
|
||||
actor.Tell(new ReconfigureConnectionCommand(name, "OpcUa", Config(NewEndpoint), null, 3));
|
||||
AwaitCondition(() => HealthTransitions(name, ConnectionHealth.Connected) >= 2,
|
||||
TimeSpan.FromSeconds(5), TimeSpan.FromMilliseconds(10));
|
||||
|
||||
gate.SetResult("sub-old");
|
||||
|
||||
var received = probe.ReceiveWhile(TimeSpan.FromSeconds(2), m => m);
|
||||
|
||||
// Positive control: the completion really did arrive and was dropped, rather than the
|
||||
// window simply having been too short to observe anything.
|
||||
Assert.Contains(received, m => m is SubscribeTagsResponse { Success: false });
|
||||
Assert.DoesNotContain(received, m => m is TagValueUpdate u && Equals(u.Value, 999));
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// A connection-level failure row belongs to the adapter that produced it. Acting on a stale
|
||||
/// one drives the healthy replacement connection into Reconnecting — dropping every tag to
|
||||
/// bad quality and re-subscribing the whole registry for nothing.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public void StaleConnectionLevelFailure_DoesNotDriveTheNewConnectionBackToReconnecting()
|
||||
{
|
||||
const string name = "stale-sub-connfail";
|
||||
var gate = new TaskCompletionSource<string>();
|
||||
|
||||
var oldAdapter = HealthyAdapter("sub-old");
|
||||
oldAdapter.SubscribeAsync(Tag, Arg.Any<SubscriptionCallback>(), Arg.Any<CancellationToken>())
|
||||
.Returns(_ => gate.Task);
|
||||
var newAdapter = HealthyAdapter("sub-new");
|
||||
ExpectNewAdapter(newAdapter);
|
||||
|
||||
var actor = CreateConnectionActor(name, oldAdapter);
|
||||
AwaitCondition(() => HealthTransitions(name, ConnectionHealth.Connected) >= 1,
|
||||
TimeSpan.FromSeconds(5), TimeSpan.FromMilliseconds(10));
|
||||
|
||||
var probe = CreateTestProbe();
|
||||
probe.Send(actor, new SubscribeTagsRequest("c1", "inst1", name, [Tag], DateTimeOffset.UtcNow));
|
||||
AwaitCondition(() => CalledWith(oldAdapter, nameof(IDataConnection.SubscribeAsync), Tag),
|
||||
TimeSpan.FromSeconds(5));
|
||||
|
||||
actor.Tell(new ReconfigureConnectionCommand(name, "OpcUa", Config(NewEndpoint), null, 3));
|
||||
AwaitCondition(() => HealthTransitions(name, ConnectionHealth.Connected) >= 2,
|
||||
TimeSpan.FromSeconds(5), TimeSpan.FromMilliseconds(10));
|
||||
|
||||
var disconnectsBefore = HealthTransitions(name, ConnectionHealth.Disconnected);
|
||||
|
||||
// InvalidOperationException is what IsConnectionLevelFailure classifies as "transport
|
||||
// down", i.e. the row that asks the state machine to reconnect.
|
||||
gate.SetException(new InvalidOperationException("adapter not connected"));
|
||||
|
||||
probe.FishForMessage<SubscribeTagsResponse>(
|
||||
m => m.CorrelationId == "c1" && !m.Success, TimeSpan.FromSeconds(5));
|
||||
|
||||
// Entering Reconnecting would push bad quality to every subscriber — the probe is one.
|
||||
probe.ExpectNoMsg(TimeSpan.FromMilliseconds(500));
|
||||
Assert.Equal(disconnectsBefore, HealthTransitions(name, ConnectionHealth.Disconnected));
|
||||
Assert.Equal(2, HealthTransitions(name, ConnectionHealth.Connected));
|
||||
}
|
||||
|
||||
// ── Alarm subscribe ──
|
||||
|
||||
/// <summary>
|
||||
/// The alarm twin. The stale completion lands first and (before the fix) stored its dead feed
|
||||
/// id, so the LIVE feed's completion took the duplicate branch and released the feed the new
|
||||
/// adapter had just established — a source that stays mirrored but never updates again.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public void StaleAlarmSubscribeCompleted_DoesNotReleaseTheLiveAlarmFeed()
|
||||
{
|
||||
const string name = "stale-alarm-feed";
|
||||
var oldGate = new TaskCompletionSource<string>();
|
||||
var newGate = new TaskCompletionSource<string>();
|
||||
AlarmTransitionCallback? newCallback = null;
|
||||
|
||||
var oldAdapter = HealthyAlarmAdapter("alarm-old");
|
||||
((IAlarmSubscribableConnection)oldAdapter)
|
||||
.SubscribeAlarmsAsync(Source, Arg.Any<string?>(),
|
||||
Arg.Any<AlarmTransitionCallback>(), Arg.Any<CancellationToken>())
|
||||
.Returns(_ => oldGate.Task);
|
||||
|
||||
var newAdapter = HealthyAlarmAdapter("alarm-new");
|
||||
((IAlarmSubscribableConnection)newAdapter)
|
||||
.SubscribeAlarmsAsync(Source, Arg.Any<string?>(),
|
||||
Arg.Do<AlarmTransitionCallback>(c => newCallback = c), Arg.Any<CancellationToken>())
|
||||
.Returns(_ => newGate.Task);
|
||||
ExpectNewAdapter(newAdapter);
|
||||
|
||||
var actor = CreateConnectionActor(name, oldAdapter);
|
||||
AwaitCondition(() => HealthTransitions(name, ConnectionHealth.Connected) >= 1,
|
||||
TimeSpan.FromSeconds(5), TimeSpan.FromMilliseconds(10));
|
||||
|
||||
var probe = CreateTestProbe();
|
||||
probe.Send(actor, new SubscribeAlarmsRequest("c1", "inst1", name, Source, null, DateTimeOffset.UtcNow));
|
||||
AwaitCondition(() => CalledWith(oldAdapter, "SubscribeAlarmsAsync", Source), TimeSpan.FromSeconds(5));
|
||||
|
||||
actor.Tell(new ReconfigureConnectionCommand(name, "OpcUa", Config(NewEndpoint), null, 3));
|
||||
|
||||
// The subscriber registry survives the swap, so the reconnect re-subscribe re-opens the
|
||||
// feed against the new adapter without the NativeAlarmActor doing anything.
|
||||
AwaitCondition(() => CalledWith(newAdapter, "SubscribeAlarmsAsync", Source),
|
||||
TimeSpan.FromSeconds(5), TimeSpan.FromMilliseconds(10));
|
||||
|
||||
// The dead feed answers first, the live one second — the ordering that made the live
|
||||
// completion look redundant.
|
||||
oldGate.SetResult("alarm-old");
|
||||
AwaitCondition(() => CalledWith(oldAdapter, "UnsubscribeAlarmsAsync", "alarm-old"),
|
||||
TimeSpan.FromSeconds(5), TimeSpan.FromMilliseconds(10));
|
||||
newGate.SetResult("alarm-new");
|
||||
|
||||
AwaitCondition(() => newCallback != null, TimeSpan.FromSeconds(5));
|
||||
// Settle (and drain the swap's expected NativeAlarmSourceUnavailable) so the live
|
||||
// completion has certainly been applied before asserting nothing released its feed.
|
||||
probe.ReceiveWhile(TimeSpan.FromMilliseconds(300), m => m);
|
||||
Assert.False(Called(newAdapter, "UnsubscribeAlarmsAsync"));
|
||||
|
||||
// And the feed is genuinely live: a transition on it still routes to the subscriber.
|
||||
newCallback!(Raise("Tank01.Hi", Source));
|
||||
probe.FishForMessage<NativeAlarmTransitionUpdate>(
|
||||
u => u.Transition.SourceObjectReference == Source, TimeSpan.FromSeconds(5));
|
||||
}
|
||||
}
|
||||
+186
@@ -0,0 +1,186 @@
|
||||
using Microsoft.Extensions.Logging.Abstractions;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Types.Alarms;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums;
|
||||
using ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Adapters;
|
||||
|
||||
namespace ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests.Adapters;
|
||||
|
||||
/// <summary>
|
||||
/// P2-2 (arch-review 2026-08-19, High): before the fix, <c>ConnectAsync</c> reset the
|
||||
/// event loop and the client but left the alarm-stream state (<c>_alarmCts</c>,
|
||||
/// <c>_alarmSubCount</c>, <c>_alarmSubSources</c>, <c>_alarmStreamPrefix</c>) untouched.
|
||||
/// After a reconnect, the DataConnectionActor's transparent re-subscribe
|
||||
/// (<c>ReSubscribeAllAlarms</c>) called <c>SubscribeAlarmsAsync</c> again, but the stale
|
||||
/// non-null <c>_alarmCts</c> plus the coverage check made it short-circuit and return
|
||||
/// success WITHOUT ever opening a new stream on the new client — leaving the orphaned
|
||||
/// old stream loop retrying the disposed old client forever and the native-alarm mirror
|
||||
/// frozen until failover/redeploy.
|
||||
/// </summary>
|
||||
[Collection("DataConnectionManagerActor")]
|
||||
public class MxGatewayDataConnectionAlarmReconnectTests
|
||||
{
|
||||
private static Dictionary<string, string> Details() => new()
|
||||
{
|
||||
["Endpoint"] = "http://gw:5000",
|
||||
["ApiKey"] = "key",
|
||||
};
|
||||
|
||||
[Fact]
|
||||
public async Task Reconnect_ThenSubscribeAlarms_OpensNewStreamOnNewClient()
|
||||
{
|
||||
var clients = new List<PerInstanceFakeMxGatewayClient>();
|
||||
var factory = new RecordingFactory(clients);
|
||||
var adapter = new MxGatewayDataConnection(factory, NullLogger<MxGatewayDataConnection>.Instance);
|
||||
|
||||
await adapter.ConnectAsync(Details());
|
||||
await adapter.SubscribeAlarmsAsync("Area1.Tank1", null, _ => { });
|
||||
await WaitUntilAsync(() => clients[0].AlarmStreamOpens.Count == 1);
|
||||
|
||||
// Reconnect (same source set as before — the coverage check would previously
|
||||
// treat this as already-covered and never touch the new client at all).
|
||||
await adapter.ConnectAsync(Details());
|
||||
Assert.Equal(2, clients.Count);
|
||||
|
||||
await adapter.SubscribeAlarmsAsync("Area1.Tank1", null, _ => { });
|
||||
|
||||
await WaitUntilAsync(() => clients[1].AlarmStreamOpens.Count == 1);
|
||||
Assert.Equal("Area1.Tank1", clients[1].AlarmStreamOpens[0]);
|
||||
// The OLD client must never see a second alarm-stream open after reconnect.
|
||||
Assert.Single(clients[0].AlarmStreamOpens);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Reconnect_CancelsTheOldAlarmStreamToken()
|
||||
{
|
||||
var clients = new List<PerInstanceFakeMxGatewayClient>();
|
||||
var factory = new RecordingFactory(clients);
|
||||
var adapter = new MxGatewayDataConnection(factory, NullLogger<MxGatewayDataConnection>.Instance);
|
||||
|
||||
await adapter.ConnectAsync(Details());
|
||||
await adapter.SubscribeAlarmsAsync("Area1.Tank1", null, _ => { });
|
||||
await WaitUntilAsync(() => clients[0].AlarmStreamTokens.Count == 1);
|
||||
var oldStreamToken = clients[0].AlarmStreamTokens[0];
|
||||
Assert.False(oldStreamToken.IsCancellationRequested);
|
||||
|
||||
await adapter.ConnectAsync(Details()); // reconnect
|
||||
|
||||
// The token the OLD RunAlarmStreamAsync loop was given must now be cancelled —
|
||||
// that is what stops it retrying against the disposed old client every 5s.
|
||||
Assert.True(oldStreamToken.IsCancellationRequested);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Reconnect_AlarmEventsFlowAgainOnTheNewClient()
|
||||
{
|
||||
var clients = new List<PerInstanceFakeMxGatewayClient>();
|
||||
var factory = new RecordingFactory(clients);
|
||||
var adapter = new MxGatewayDataConnection(factory, NullLogger<MxGatewayDataConnection>.Instance);
|
||||
|
||||
var received = new List<NativeAlarmTransition>();
|
||||
await adapter.ConnectAsync(Details());
|
||||
await adapter.SubscribeAlarmsAsync("Area1.Tank1", null, t => received.Add(t));
|
||||
await WaitUntilAsync(() => clients[0].AlarmStreamOpens.Count == 1);
|
||||
|
||||
// Prove the feed works pre-reconnect.
|
||||
clients[0].EmitAlarm(MakeTransition(AlarmTransitionKind.Snapshot));
|
||||
await WaitUntilAsync(() => received.Count == 1);
|
||||
|
||||
await adapter.ConnectAsync(Details()); // reconnect
|
||||
await adapter.SubscribeAlarmsAsync("Area1.Tank1", null, t => received.Add(t));
|
||||
await WaitUntilAsync(() => clients[1].AlarmStreamOpens.Count == 1);
|
||||
|
||||
// A transition on the OLD (disposed, orphaned) client must never reach the
|
||||
// callback again — only the NEW client's feed is live post-reconnect.
|
||||
clients[0].EmitAlarm(MakeTransition(AlarmTransitionKind.Raise));
|
||||
await Task.Delay(100);
|
||||
Assert.Single(received);
|
||||
|
||||
clients[1].EmitAlarm(MakeTransition(AlarmTransitionKind.Snapshot));
|
||||
await WaitUntilAsync(() => received.Count == 2);
|
||||
}
|
||||
|
||||
private static NativeAlarmTransition MakeTransition(AlarmTransitionKind kind) => new(
|
||||
"Area1.Tank1.HiAlarm", "Area1.Tank1", "HiAlarm", kind,
|
||||
new AlarmConditionState(true, false, null, AlarmShelveState.Unshelved, false, 400),
|
||||
"Area", "", "", "", "", null, DateTimeOffset.UtcNow, "", "");
|
||||
|
||||
private static async Task WaitUntilAsync(Func<bool> condition)
|
||||
{
|
||||
for (var i = 0; i < 100 && !condition(); i++)
|
||||
await Task.Delay(50);
|
||||
Assert.True(condition(), "condition not met within timeout");
|
||||
}
|
||||
|
||||
/// <summary>Builds a NEW client instance per <c>Create()</c> call, mirroring the real
|
||||
/// factory and <see cref="MxGatewayDataConnectionReconnectTests"/>'s NSubstitute-based
|
||||
/// setup — needed here (unlike <c>FakeMxGatewayClient</c>) because the test must tell
|
||||
/// the pre- and post-reconnect clients apart.</summary>
|
||||
private sealed class RecordingFactory : IMxGatewayClientFactory
|
||||
{
|
||||
private readonly List<PerInstanceFakeMxGatewayClient> _clients;
|
||||
public RecordingFactory(List<PerInstanceFakeMxGatewayClient> clients) => _clients = clients;
|
||||
|
||||
public IMxGatewayClient Create()
|
||||
{
|
||||
var c = new PerInstanceFakeMxGatewayClient();
|
||||
_clients.Add(c);
|
||||
return c;
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>Minimal per-instance fake — only the members this test exercises are
|
||||
/// implemented meaningfully; the rest are unused no-ops for the connect/alarm-only
|
||||
/// path under test.</summary>
|
||||
private sealed class PerInstanceFakeMxGatewayClient : IMxGatewayClient
|
||||
{
|
||||
public readonly List<string?> AlarmStreamOpens = new();
|
||||
public readonly List<CancellationToken> AlarmStreamTokens = new();
|
||||
private Action<NativeAlarmTransition>? _onTransition;
|
||||
private CancellationToken _streamToken;
|
||||
|
||||
public Task ConnectAsync(MxGatewayConnectionOptions o, CancellationToken ct = default) => Task.CompletedTask;
|
||||
public Task DisconnectAsync(CancellationToken ct = default) => Task.CompletedTask;
|
||||
public Task<string> SubscribeAsync(string tagPath, CancellationToken ct = default) => Task.FromResult(Guid.NewGuid().ToString());
|
||||
public Task UnsubscribeAsync(string subscriptionId, CancellationToken ct = default) => Task.CompletedTask;
|
||||
public Task<IReadOnlyList<MxSubscribeOutcome>> SubscribeBulkAsync(IReadOnlyList<string> tagPaths, CancellationToken ct = default)
|
||||
=> Task.FromResult<IReadOnlyList<MxSubscribeOutcome>>(Array.Empty<MxSubscribeOutcome>());
|
||||
public Task UnsubscribeBulkAsync(IReadOnlyList<string> subscriptionIds, CancellationToken ct = default) => Task.CompletedTask;
|
||||
public Task<IReadOnlyList<MxReadOutcome>> ReadAsync(IReadOnlyList<string> tagPaths, CancellationToken ct = default)
|
||||
=> Task.FromResult<IReadOnlyList<MxReadOutcome>>(Array.Empty<MxReadOutcome>());
|
||||
public Task<IReadOnlyList<MxWriteOutcome>> WriteAsync(IReadOnlyList<(string TagPath, object? Value)> writes, CancellationToken ct = default)
|
||||
=> Task.FromResult<IReadOnlyList<MxWriteOutcome>>(Array.Empty<MxWriteOutcome>());
|
||||
public Task<(IReadOnlyList<MxBrowseChild> Children, bool Truncated)> BrowseChildrenAsync(string? parentNodeId, CancellationToken ct = default)
|
||||
=> Task.FromResult<(IReadOnlyList<MxBrowseChild>, bool)>((Array.Empty<MxBrowseChild>(), false));
|
||||
public Task RunEventLoopAsync(Action<MxValueUpdate> onUpdate, CancellationToken ct = default)
|
||||
=> Task.Delay(Timeout.Infinite, ct).ContinueWith(_ => { }, TaskScheduler.Default);
|
||||
|
||||
public Task RunAlarmStreamAsync(string? alarmFilterPrefix, Action<NativeAlarmTransition> onTransition, CancellationToken ct = default)
|
||||
{
|
||||
// Mirrors the real RunAlarmStreamAsync seam: records the open, then just
|
||||
// parks on the token like the live gateway stream would between events —
|
||||
// the adapter-level bug under test is about which CLIENT/TOKEN the loop was
|
||||
// started on, not the internal 5s retry loop inside RealMxGatewayClient.
|
||||
lock (AlarmStreamOpens)
|
||||
{
|
||||
AlarmStreamOpens.Add(alarmFilterPrefix);
|
||||
AlarmStreamTokens.Add(ct);
|
||||
}
|
||||
_onTransition = onTransition;
|
||||
_streamToken = ct;
|
||||
return Task.Delay(Timeout.Infinite, ct).ContinueWith(_ => { }, TaskScheduler.Default);
|
||||
}
|
||||
|
||||
public ValueTask DisposeAsync() => ValueTask.CompletedTask;
|
||||
|
||||
/// <summary>Simulates a transition arriving on this client's live stream. Gated on
|
||||
/// the stream token exactly like the real gateway stream would be — once cancelled,
|
||||
/// no more events are delivered — so a test can prove the OLD client's feed is
|
||||
/// genuinely dead post-reconnect rather than just unobserved.</summary>
|
||||
public void EmitAlarm(NativeAlarmTransition transition)
|
||||
{
|
||||
if (_streamToken.IsCancellationRequested)
|
||||
return;
|
||||
_onTransition?.Invoke(transition);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -68,6 +68,19 @@ public sealed record HarnessConfig
|
||||
/// </summary>
|
||||
public int StreamProbesPerSite { get; init; } = 10;
|
||||
|
||||
/// <summary>
|
||||
/// Instances per site additionally watched over a REAL gRPC hop: a per-site
|
||||
/// Kestrel h2c listener hosts the production <c>SiteStreamGrpcServer</c> and the
|
||||
/// production <c>SiteStreamGrpcClient</c> dials in with batching negotiated, so
|
||||
/// the R2 <c>SiteStreamEventBatcher</c>, protobuf serialization and HTTP/2 flow
|
||||
/// control are inside the measured path (they are NOT for
|
||||
/// <see cref="StreamProbesPerSite"/>, whose final hop is deliberately an
|
||||
/// in-process reader). 0 disables the wire path entirely. Like the in-process
|
||||
/// probes, each wire subscriber's stream graph sees the full site event flow and
|
||||
/// filters it, so this adds to the fan-out multiplier.
|
||||
/// </summary>
|
||||
public int WireProbesPerSite { get; init; } = 10;
|
||||
|
||||
/// <summary>Store-and-forward messages enqueued for the drain-rate measurement (register row 50).</summary>
|
||||
public int StoreAndForwardDrainMessages { get; init; } = 20_000;
|
||||
|
||||
@@ -131,6 +144,7 @@ public sealed record HarnessConfig
|
||||
"health-interval-seconds" => config with { HealthReportInterval = TimeSpan.FromSeconds(double.Parse(value)) },
|
||||
"debug-probe-interval-seconds" => config with { DebugProbeInterval = TimeSpan.FromSeconds(double.Parse(value)) },
|
||||
"stream-probes-per-site" => config with { StreamProbesPerSite = int.Parse(value) },
|
||||
"wire-probes-per-site" => config with { WireProbesPerSite = int.Parse(value) },
|
||||
"sf-drain-messages" => config with { StoreAndForwardDrainMessages = int.Parse(value) },
|
||||
"slow-subscriber-events" => config with { SlowSubscriberEvents = int.Parse(value) },
|
||||
"data-dir" => config with { DataDirectory = value },
|
||||
|
||||
@@ -40,6 +40,10 @@ namespace ZB.MOM.WW.ScadaBridge.LoadHarness;
|
||||
/// <param name="StreamProbeDropped">Events evicted by live subscribers' bounded channels.</param>
|
||||
/// <param name="StoreAndForwardDrain">Store-and-forward drain measurement (register row 50).</param>
|
||||
/// <param name="SlowSubscriber">Slow-subscriber isolation measurement (register row 50).</param>
|
||||
/// <param name="WireStream">
|
||||
/// Wire-path stream measurement — real Kestrel + real <c>SiteStreamGrpcClient</c> with
|
||||
/// R2 batching negotiated — or null when <see cref="HarnessConfig.WireProbesPerSite"/> is 0.
|
||||
/// </param>
|
||||
public sealed record HarnessRunResult(
|
||||
HarnessConfig Config,
|
||||
EnvironmentInfo Environment,
|
||||
@@ -67,7 +71,30 @@ public sealed record HarnessRunResult(
|
||||
long StreamProbeReceived,
|
||||
long StreamProbeDropped,
|
||||
StoreAndForwardDrainResult? StoreAndForwardDrain,
|
||||
SlowSubscriberResult? SlowSubscriber);
|
||||
SlowSubscriberResult? SlowSubscriber,
|
||||
WireStreamResult? WireStream);
|
||||
|
||||
/// <summary>
|
||||
/// Steady-state-window measurement of the wire-path stream probes: the production
|
||||
/// gRPC server (Kestrel h2c) and client with the R2 batcher, protobuf serialization
|
||||
/// and HTTP/2 flow control INSIDE the measured path. All figures are deltas over the
|
||||
/// steady-state window only.
|
||||
/// </summary>
|
||||
/// <param name="ProbeCount">Wire subscribers attached across the fleet.</param>
|
||||
/// <param name="Latency">End-to-end DCL-boundary → post-unpack client callback latency.</param>
|
||||
/// <param name="EventsReceived">Domain events delivered by the client in the window.</param>
|
||||
/// <param name="ServerDroppedEvents">Events evicted by the wire servers' bounded send channels in the window.</param>
|
||||
/// <param name="StreamFaults">Stream faults reported by the client (expected 0).</param>
|
||||
/// <param name="GracefulCompletions">Graceful server-side stream completions (expected 0 inside the 4 h lifetime).</param>
|
||||
/// <param name="Batches">The frame/batch-size distribution the R2 batcher actually produced.</param>
|
||||
public sealed record WireStreamResult(
|
||||
int ProbeCount,
|
||||
LatencySnapshot Latency,
|
||||
long EventsReceived,
|
||||
long ServerDroppedEvents,
|
||||
long StreamFaults,
|
||||
long GracefulCompletions,
|
||||
BatchSizeWindow Batches);
|
||||
|
||||
/// <summary>Host and runtime facts recorded alongside the numbers.</summary>
|
||||
/// <param name="MachineName">Host name.</param>
|
||||
@@ -120,6 +147,7 @@ public static class HarnessRun
|
||||
ObservabilityProbes? probes = null;
|
||||
StoreAndForwardDrainResult? drainResult = null;
|
||||
SlowSubscriberResult? slowResult = null;
|
||||
BatchSizeAggregator? batchAggregator = null;
|
||||
|
||||
var aggregator = new CentralHealthAggregator(
|
||||
Options.Create(new HealthMonitoringOptions
|
||||
@@ -154,6 +182,18 @@ public static class HarnessRun
|
||||
site.AttachStreamProbes(latency);
|
||||
log($" {sites.Sum(s => s.Probes.Count)} live stream subscribers attached");
|
||||
|
||||
// Wire path (real gRPC): the aggregator must exist BEFORE the first stream
|
||||
// opens or the MeterListener misses early frames.
|
||||
var wireLatency = new LatencyHistogram();
|
||||
if (config.WireProbesPerSite > 0)
|
||||
{
|
||||
batchAggregator = new BatchSizeAggregator();
|
||||
foreach (var site in sites)
|
||||
await site.AttachWireStreamProbesAsync(config.WireProbesPerSite, wireLatency, log);
|
||||
log($" {sites.Sum(s => s.WireSubscribers.Count)} wire stream subscribers attached " +
|
||||
"(real Kestrel + SiteStreamGrpcClient, batching negotiated)");
|
||||
}
|
||||
|
||||
// Let Instance Actors complete their DCL subscribe round-trips before the
|
||||
// driver starts; an emit before SubscribeBatchAsync has captured the
|
||||
// callback would be silently discarded.
|
||||
@@ -184,6 +224,18 @@ public static class HarnessRun
|
||||
probe.RetargetLatency(steadyLatency);
|
||||
}
|
||||
|
||||
// Wire path: same ramp/steady histogram separation, plus window-start
|
||||
// baselines for the client/server counters and the batch distribution.
|
||||
var steadyWireLatency = new LatencyHistogram();
|
||||
var wireReceivedAtWindowStart = sites.Sum(s => s.WireSubscribers.Sum(p => p.ReceivedEvents));
|
||||
var wireDroppedAtWindowStart = sites.Sum(s => s.WireHost?.Server.DroppedStreamEventCount ?? 0);
|
||||
var batchesAtWindowStart = batchAggregator?.Snapshot();
|
||||
foreach (var site in sites)
|
||||
{
|
||||
foreach (var wire in site.WireSubscribers)
|
||||
wire.RetargetLatency(steadyWireLatency);
|
||||
}
|
||||
|
||||
var windowWatch = Stopwatch.StartNew();
|
||||
log($"Sustained measurement window: {config.SustainDuration.TotalMinutes:F1} min...");
|
||||
await Task.Delay(config.SustainDuration, cancellationToken);
|
||||
@@ -199,6 +251,24 @@ public static class HarnessRun
|
||||
var probeReceived = sites.Sum(s => s.Probes.Sum(p => p.ReceivedEvents)) - receivedAtWindowStart;
|
||||
var probeDropped = sites.Sum(s => s.Probes.Sum(p => p.DroppedEvents)) - droppedAtWindowStart;
|
||||
|
||||
// Wire-path window deltas, captured BEFORE the row-50 scenarios below so
|
||||
// their publishing cannot contaminate the wire figures.
|
||||
WireStreamResult? wireResult = null;
|
||||
if (batchAggregator is not null)
|
||||
{
|
||||
wireResult = new WireStreamResult(
|
||||
ProbeCount: sites.Sum(s => s.WireSubscribers.Count),
|
||||
Latency: steadyWireLatency.Snapshot(),
|
||||
EventsReceived: sites.Sum(s => s.WireSubscribers.Sum(p => p.ReceivedEvents)) - wireReceivedAtWindowStart,
|
||||
ServerDroppedEvents: sites.Sum(s => s.WireHost?.Server.DroppedStreamEventCount ?? 0) - wireDroppedAtWindowStart,
|
||||
StreamFaults: sites.Sum(s => s.WireSubscribers.Sum(p => p.Faults)),
|
||||
GracefulCompletions: sites.Sum(s => s.WireSubscribers.Sum(p => p.GracefulCompletions)),
|
||||
Batches: BatchSizeWindow.From(batchesAtWindowStart!, batchAggregator.Snapshot()));
|
||||
log($" wire path: {wireResult.EventsReceived:N0} events in {wireResult.Batches.Frames:N0} frames " +
|
||||
$"(mean {wireResult.Batches.MeanEventsPerFrame:F1} ev/frame, max {wireResult.Batches.Max}), " +
|
||||
$"P99 {wireResult.Latency.P99Ms:F2}ms");
|
||||
}
|
||||
|
||||
// Register row 50 — measured after the sustained window so the numbers are
|
||||
// not competing with the full tag load for CPU, and reported separately for
|
||||
// the same reason.
|
||||
@@ -250,12 +320,14 @@ public static class HarnessRun
|
||||
StreamProbeReceived: probeReceived,
|
||||
StreamProbeDropped: probeDropped,
|
||||
StoreAndForwardDrain: drainResult,
|
||||
SlowSubscriber: slowResult);
|
||||
SlowSubscriber: slowResult,
|
||||
WireStream: wireResult);
|
||||
}
|
||||
finally
|
||||
{
|
||||
if (driver != null) await driver.DisposeAsync();
|
||||
if (probes != null) await probes.DisposeAsync();
|
||||
batchAggregator?.Dispose();
|
||||
await sampler.DisposeAsync();
|
||||
|
||||
log("Tearing down sites...");
|
||||
|
||||
@@ -0,0 +1,136 @@
|
||||
using System.Diagnostics.Metrics;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Observability;
|
||||
|
||||
namespace ZB.MOM.WW.ScadaBridge.LoadHarness.Metrics;
|
||||
|
||||
/// <summary>
|
||||
/// Captures the production <c>scadabridge.site.stream.batch_size</c> histogram via a
|
||||
/// <see cref="MeterListener"/>, so a wire-path run can report the actual frame/batch
|
||||
/// distribution the R2 batcher produced — not a client-side reconstruction. The
|
||||
/// instrument is recorded once per emitted gRPC frame (negotiated streams only), with
|
||||
/// the frame's event count as the value, which makes <c>count = frames</c> and
|
||||
/// <c>sum = events</c> exactly.
|
||||
///
|
||||
/// <para>
|
||||
/// The harness hosts every site's <see cref="ZB.MOM.WW.ScadaBridge.Communication.Grpc.SiteStreamGrpcServer"/>
|
||||
/// in-process, so this listener sees every wire frame the run emits and nothing else —
|
||||
/// no other component in the harness records this instrument.
|
||||
/// </para>
|
||||
/// </summary>
|
||||
public sealed class BatchSizeAggregator : IDisposable
|
||||
{
|
||||
/// <summary>
|
||||
/// Sizes above this are clamped into the last bucket. The production cap is
|
||||
/// <c>GrpcStreamBatchMaxEvents</c> = 100, so 512 leaves generous headroom for a
|
||||
/// configured-up run without an unbounded array.
|
||||
/// </summary>
|
||||
public const int MaxTrackedBatchSize = 512;
|
||||
|
||||
private readonly MeterListener _listener;
|
||||
private readonly long[] _framesBySize = new long[MaxTrackedBatchSize + 1];
|
||||
|
||||
/// <summary>Starts listening. Must be created BEFORE the first stream opens.</summary>
|
||||
public BatchSizeAggregator()
|
||||
{
|
||||
_listener = new MeterListener
|
||||
{
|
||||
InstrumentPublished = (instrument, listener) =>
|
||||
{
|
||||
if (instrument.Meter.Name == ScadaBridgeTelemetry.MeterName &&
|
||||
instrument.Name == "scadabridge.site.stream.batch_size")
|
||||
{
|
||||
listener.EnableMeasurementEvents(instrument);
|
||||
}
|
||||
},
|
||||
};
|
||||
_listener.SetMeasurementEventCallback<int>((_, value, _, _) =>
|
||||
{
|
||||
var size = Math.Clamp(value, 1, MaxTrackedBatchSize);
|
||||
Interlocked.Increment(ref _framesBySize[size]);
|
||||
});
|
||||
_listener.Start();
|
||||
}
|
||||
|
||||
/// <summary>Copies the cumulative per-size frame counts.</summary>
|
||||
/// <returns>Index = events-per-frame (clamped), value = frames emitted at that size.</returns>
|
||||
public long[] Snapshot()
|
||||
{
|
||||
var copy = new long[_framesBySize.Length];
|
||||
for (var i = 0; i < copy.Length; i++)
|
||||
copy[i] = Interlocked.Read(ref _framesBySize[i]);
|
||||
return copy;
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public void Dispose() => _listener.Dispose();
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Batch-size distribution over one measurement window, computed from two
|
||||
/// <see cref="BatchSizeAggregator.Snapshot"/>s. Percentiles are exact (integer sizes,
|
||||
/// full counts — nothing is bucketed away).
|
||||
/// </summary>
|
||||
/// <param name="Frames">gRPC frames written in the window.</param>
|
||||
/// <param name="Events">Events those frames carried.</param>
|
||||
/// <param name="BatchedFrames">Frames that carried more than one event.</param>
|
||||
/// <param name="MeanEventsPerFrame">Events / Frames.</param>
|
||||
/// <param name="P50">Median events-per-frame.</param>
|
||||
/// <param name="P95">95th percentile events-per-frame.</param>
|
||||
/// <param name="P99">99th percentile events-per-frame.</param>
|
||||
/// <param name="Max">Largest frame observed in the window.</param>
|
||||
public sealed record BatchSizeWindow(
|
||||
long Frames,
|
||||
long Events,
|
||||
long BatchedFrames,
|
||||
double MeanEventsPerFrame,
|
||||
int P50,
|
||||
int P95,
|
||||
int P99,
|
||||
int Max)
|
||||
{
|
||||
/// <summary>Computes the window distribution from cumulative start/end snapshots.</summary>
|
||||
/// <param name="start">Snapshot at window start.</param>
|
||||
/// <param name="end">Snapshot at window end.</param>
|
||||
/// <returns>The distribution of frames emitted between the two snapshots.</returns>
|
||||
public static BatchSizeWindow From(long[] start, long[] end)
|
||||
{
|
||||
long frames = 0, events = 0, batched = 0;
|
||||
var max = 0;
|
||||
var delta = new long[end.Length];
|
||||
for (var size = 1; size < end.Length; size++)
|
||||
{
|
||||
var n = end[size] - start[size];
|
||||
if (n <= 0) continue;
|
||||
delta[size] = n;
|
||||
frames += n;
|
||||
events += n * size;
|
||||
if (size > 1) batched += n;
|
||||
max = size;
|
||||
}
|
||||
|
||||
return new BatchSizeWindow(
|
||||
frames, events, batched,
|
||||
frames == 0 ? 0 : events / (double)frames,
|
||||
Percentile(delta, frames, 50),
|
||||
Percentile(delta, frames, 95),
|
||||
Percentile(delta, frames, 99),
|
||||
max);
|
||||
}
|
||||
|
||||
private static int Percentile(long[] delta, long frames, double percentile)
|
||||
{
|
||||
if (frames == 0) return 0;
|
||||
var target = (long)Math.Ceiling(frames * percentile / 100.0);
|
||||
if (target < 1) target = 1;
|
||||
|
||||
long cumulative = 0;
|
||||
for (var size = 1; size < delta.Length; size++)
|
||||
{
|
||||
cumulative += delta[size];
|
||||
if (cumulative >= target)
|
||||
return size;
|
||||
}
|
||||
|
||||
return delta.Length - 1;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,96 @@
|
||||
using Microsoft.AspNetCore.Builder;
|
||||
using Microsoft.AspNetCore.Hosting;
|
||||
using Microsoft.AspNetCore.Hosting.Server;
|
||||
using Microsoft.AspNetCore.Hosting.Server.Features;
|
||||
using Microsoft.AspNetCore.Server.Kestrel.Core;
|
||||
using Microsoft.Extensions.DependencyInjection;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using Microsoft.Extensions.Options;
|
||||
using ZB.MOM.WW.ScadaBridge.Communication;
|
||||
using ZB.MOM.WW.ScadaBridge.Communication.Grpc;
|
||||
|
||||
namespace ZB.MOM.WW.ScadaBridge.LoadHarness.Probes;
|
||||
|
||||
/// <summary>
|
||||
/// The site half of the wire-path stream probe: a real Kestrel HTTP/2 (h2c) listener on
|
||||
/// a loopback ephemeral port, serving the PRODUCTION <see cref="SiteStreamGrpcServer"/>
|
||||
/// over the site's real <see cref="ZB.MOM.WW.ScadaBridge.SiteRuntime.Streaming.SiteStreamManager"/>.
|
||||
///
|
||||
/// <para>
|
||||
/// This is the hop <see cref="StreamSubscriberProbe"/> deliberately substitutes with a
|
||||
/// controllable in-process reader (which the slow-subscriber scenario needs), and
|
||||
/// therefore the hop the original WP-4 runs never exercised: the
|
||||
/// <c>SiteStreamEventBatcher</c> pump (R2), protobuf serialization,
|
||||
/// <c>IServerStreamWriter.WriteAsync</c>, and HTTP/2 flow control. The server is built
|
||||
/// through its production DI constructor with default <see cref="CommunicationOptions"/>,
|
||||
/// so the batch negotiation (100 events / 25 ms window), channel capacities and the
|
||||
/// concurrency cap are exactly what ships.
|
||||
/// </para>
|
||||
///
|
||||
/// <para>
|
||||
/// Deliberately NOT in the path: <c>ControlPlaneAuthInterceptor</c> (it lives in the
|
||||
/// Host project and authenticates once per CALL — one header compare per stream
|
||||
/// lifetime, nothing on the per-event path) and TLS (production is h2c too).
|
||||
/// </para>
|
||||
/// </summary>
|
||||
public sealed class SiteWireStreamHost : IAsyncDisposable
|
||||
{
|
||||
private readonly WebApplication _app;
|
||||
|
||||
/// <summary>The hosted production gRPC server (drop counters, active stream count).</summary>
|
||||
public SiteStreamGrpcServer Server { get; }
|
||||
|
||||
/// <summary>The dialable endpoint, e.g. <c>http://127.0.0.1:54321</c>.</summary>
|
||||
public string Endpoint { get; }
|
||||
|
||||
private SiteWireStreamHost(WebApplication app, SiteStreamGrpcServer server, string endpoint)
|
||||
{
|
||||
_app = app;
|
||||
Server = server;
|
||||
Endpoint = endpoint;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Builds, starts and readies the listener for one site.
|
||||
/// </summary>
|
||||
/// <param name="site">The site whose stream manager and actor system back the server.</param>
|
||||
/// <returns>The started host.</returns>
|
||||
public static async Task<SiteWireStreamHost> StartAsync(SiteRuntimeFixture site)
|
||||
{
|
||||
var server = new SiteStreamGrpcServer(
|
||||
site.StreamManager,
|
||||
Microsoft.Extensions.Logging.Abstractions.NullLogger<SiteStreamGrpcServer>.Instance,
|
||||
Options.Create(new CommunicationOptions()));
|
||||
|
||||
var builder = WebApplication.CreateSlimBuilder();
|
||||
builder.Logging.ClearProviders();
|
||||
builder.WebHost.ConfigureKestrel(options =>
|
||||
options.Listen(System.Net.IPAddress.Loopback, 0, listenOptions =>
|
||||
listenOptions.Protocols = HttpProtocols.Http2));
|
||||
builder.Services.AddGrpc();
|
||||
builder.Services.AddSingleton(server);
|
||||
|
||||
var app = builder.Build();
|
||||
app.MapGrpcService<SiteStreamGrpcServer>();
|
||||
await app.StartAsync();
|
||||
|
||||
// Same post-start readiness handoff the production host performs.
|
||||
server.SetReady(site.System);
|
||||
|
||||
var address = app.Services.GetRequiredService<IServer>()
|
||||
.Features.Get<IServerAddressesFeature>()?.Addresses.FirstOrDefault()
|
||||
?? throw new InvalidOperationException("Kestrel bound no address.");
|
||||
|
||||
return new SiteWireStreamHost(app, server, address);
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public async ValueTask DisposeAsync()
|
||||
{
|
||||
// Mirrors the production shutdown ordering: cancel active streams first so
|
||||
// their pumps observe cancellation, then stop the listener.
|
||||
Server.CancelAllStreams();
|
||||
await _app.StopAsync();
|
||||
await _app.DisposeAsync();
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,108 @@
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Streaming;
|
||||
using ZB.MOM.WW.ScadaBridge.Communication.Grpc;
|
||||
using ZB.MOM.WW.ScadaBridge.LoadHarness.Metrics;
|
||||
|
||||
namespace ZB.MOM.WW.ScadaBridge.LoadHarness.Probes;
|
||||
|
||||
/// <summary>
|
||||
/// The central half of the wire-path stream probe: one instance subscription opened
|
||||
/// through the PRODUCTION <see cref="SiteStreamGrpcClient"/> — the exact
|
||||
/// <c>SubscribeAsync</c> path central's <c>DebugStreamBridgeActor</c> uses, including
|
||||
/// the R2 batch negotiation (<c>BatchingSupported = true</c>) and the client-side
|
||||
/// batch unpack + proto→domain conversion.
|
||||
///
|
||||
/// <para>
|
||||
/// Latency is recorded in <c>onEvent</c>, i.e. AFTER wire deserialization, batch
|
||||
/// unpacking and domain mapping — the same "DCL boundary → subscriber" measurement as
|
||||
/// <see cref="StreamSubscriberProbe"/>, extended across the real socket. The emit
|
||||
/// instant travels verbatim on the proto <c>Timestamp</c>, so the subtraction stays a
|
||||
/// genuine end-to-end figure (same process, same clock).
|
||||
/// </para>
|
||||
/// </summary>
|
||||
public sealed class WireStreamSubscriber
|
||||
{
|
||||
private Task _streamTask = Task.CompletedTask;
|
||||
private volatile LatencyHistogram? _latency;
|
||||
private long _received;
|
||||
private long _faults;
|
||||
private long _completions;
|
||||
|
||||
/// <summary>Probe name (also the stream's correlation id).</summary>
|
||||
public string Name { get; }
|
||||
|
||||
/// <summary>Domain events delivered by the client callback.</summary>
|
||||
public long ReceivedEvents => Interlocked.Read(ref _received);
|
||||
|
||||
/// <summary>Stream faults reported via <c>onError</c>. Expected 0 for a healthy run.</summary>
|
||||
public long Faults => Interlocked.Read(ref _faults);
|
||||
|
||||
/// <summary>Graceful server-side completions via <c>onCompleted</c>. Expected 0 inside a run shorter than the 4 h stream lifetime.</summary>
|
||||
public long GracefulCompletions => Interlocked.Read(ref _completions);
|
||||
|
||||
private WireStreamSubscriber(string name, LatencyHistogram? latency)
|
||||
{
|
||||
Name = name;
|
||||
_latency = latency;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Opens one instance stream on the shared per-site client. The subscription task
|
||||
/// runs until <paramref name="ct"/> cancels — the harness deliberately does NOT
|
||||
/// reconnect on fault (a fault inside a 25-minute loopback run is a finding, not
|
||||
/// an operational event to paper over), it counts it.
|
||||
/// </summary>
|
||||
/// <param name="client">The site's shared production gRPC client (one channel per site, streams multiplexed).</param>
|
||||
/// <param name="instanceUniqueName">Instance whose events this probe receives.</param>
|
||||
/// <param name="name">Probe/correlation name (must be a valid Akka path element).</param>
|
||||
/// <param name="latency">Histogram fed with end-to-end event latency.</param>
|
||||
/// <param name="log">Progress sink for fault reporting.</param>
|
||||
/// <param name="ct">Stops the subscription.</param>
|
||||
/// <returns>The attached subscriber.</returns>
|
||||
public static WireStreamSubscriber Attach(
|
||||
SiteStreamGrpcClient client,
|
||||
string instanceUniqueName,
|
||||
string name,
|
||||
LatencyHistogram? latency,
|
||||
Action<string> log,
|
||||
CancellationToken ct)
|
||||
{
|
||||
var self = new WireStreamSubscriber(name, latency);
|
||||
|
||||
self._streamTask = client.SubscribeAsync(
|
||||
correlationId: name,
|
||||
instanceUniqueName: instanceUniqueName,
|
||||
onEvent: evt =>
|
||||
{
|
||||
Interlocked.Increment(ref self._received);
|
||||
if (self._latency is { } histogram && evt is AttributeValueChanged changed)
|
||||
histogram.Record(DateTimeOffset.UtcNow - changed.Timestamp);
|
||||
},
|
||||
onError: ex =>
|
||||
{
|
||||
Interlocked.Increment(ref self._faults);
|
||||
log($" wire probe {name} stream fault: {ex.GetType().Name}: {ex.Message}");
|
||||
},
|
||||
onCompleted: () => Interlocked.Increment(ref self._completions),
|
||||
ct);
|
||||
|
||||
return self;
|
||||
}
|
||||
|
||||
/// <summary>Repoints the latency histogram (steady-state window separation) — mirrors <see cref="StreamSubscriberProbe.RetargetLatency"/>.</summary>
|
||||
/// <param name="latency">The histogram to record into from now on, or null to stop recording.</param>
|
||||
public void RetargetLatency(LatencyHistogram? latency) => _latency = latency;
|
||||
|
||||
/// <summary>Awaits the subscription task after its token has been cancelled.</summary>
|
||||
/// <returns>A task that completes when the stream loop has exited.</returns>
|
||||
public async Task WaitForShutdownAsync()
|
||||
{
|
||||
try
|
||||
{
|
||||
await _streamTask;
|
||||
}
|
||||
catch (OperationCanceledException)
|
||||
{
|
||||
// Normal teardown.
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -22,6 +22,7 @@ Console.WriteLine($" sites {config.Sites}");
|
||||
Console.WriteLine($" instances/site {config.InstancesPerSite}");
|
||||
Console.WriteLine($" tags/instance {config.TagsPerInstance}");
|
||||
Console.WriteLine($" total subscriptions {config.TotalSubscriptions:N0}");
|
||||
Console.WriteLine($" stream probes/site {config.StreamProbesPerSite} in-process + {config.WireProbesPerSite} wire (real gRPC)");
|
||||
Console.WriteLine($" nominal update rate {config.NominalUpdatesPerSecond:N0}/s");
|
||||
Console.WriteLine($" settle / sustain {config.SettleDuration.TotalMinutes:F1} / {config.SustainDuration.TotalMinutes:F1} min");
|
||||
Console.WriteLine();
|
||||
|
||||
@@ -47,6 +47,18 @@ public static class ResultsFormatter
|
||||
builder.AppendLine(culture, $"events dropped {result.StreamProbeDropped:N0}");
|
||||
builder.AppendLine();
|
||||
|
||||
if (result.WireStream is { } wire)
|
||||
{
|
||||
builder.AppendLine("-- wire stream subscribers (real gRPC, R2 batching in path) --");
|
||||
builder.AppendLine(culture, $"probes {wire.ProbeCount} (Kestrel h2c + SiteStreamGrpcClient, batching negotiated)");
|
||||
AppendLatency(builder, culture, wire.Latency);
|
||||
builder.AppendLine(culture, $"events delivered {wire.EventsReceived:N0} in {wire.Batches.Frames:N0} frames");
|
||||
builder.AppendLine(culture, $"batching mean {wire.Batches.MeanEventsPerFrame:F2} ev/frame p50 {wire.Batches.P50} p95 {wire.Batches.P95} p99 {wire.Batches.P99} max {wire.Batches.Max} ({wire.Batches.BatchedFrames:N0} frames >1 event)");
|
||||
builder.AppendLine(culture, $"server channel drops {wire.ServerDroppedEvents:N0}");
|
||||
builder.AppendLine(culture, $"faults / completions {wire.StreamFaults:N0} / {wire.GracefulCompletions:N0}");
|
||||
builder.AppendLine();
|
||||
}
|
||||
|
||||
builder.AppendLine("-- health report delivery --");
|
||||
AppendLatency(builder, culture, result.HealthReportLatency);
|
||||
builder.AppendLine(culture, $"reports ingested {result.HealthReportsDelivered:N0}");
|
||||
|
||||
@@ -6,6 +6,7 @@ using Microsoft.Extensions.DependencyInjection;
|
||||
using Microsoft.Extensions.Logging.Abstractions;
|
||||
using ZB.MOM.WW.LocalDb;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Messages.DataConnection;
|
||||
using ZB.MOM.WW.ScadaBridge.Communication.Grpc;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Types.Flattening;
|
||||
using ZB.MOM.WW.ScadaBridge.DataConnectionLayer;
|
||||
using ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Actors;
|
||||
@@ -43,7 +44,11 @@ public sealed class SiteRuntimeFixture : IAsyncDisposable
|
||||
private readonly ILocalDb _localDb;
|
||||
private readonly List<IActorRef> _instanceActors = new();
|
||||
private readonly List<StreamSubscriberProbe> _probes = new();
|
||||
private readonly List<WireStreamSubscriber> _wireSubscribers = new();
|
||||
private readonly string _dataDirectory;
|
||||
private SiteWireStreamHost? _wireHost;
|
||||
private SiteStreamGrpcClient? _wireClient;
|
||||
private CancellationTokenSource? _wireCts;
|
||||
|
||||
/// <summary>Site identifier, e.g. <c>site-01</c>.</summary>
|
||||
public string SiteId { get; }
|
||||
@@ -72,6 +77,12 @@ public sealed class SiteRuntimeFixture : IAsyncDisposable
|
||||
/// <summary>The live stream subscribers attached to this site.</summary>
|
||||
public IReadOnlyList<StreamSubscriberProbe> Probes => _probes;
|
||||
|
||||
/// <summary>The wire-path (real gRPC) subscribers attached to this site, if any.</summary>
|
||||
public IReadOnlyList<WireStreamSubscriber> WireSubscribers => _wireSubscribers;
|
||||
|
||||
/// <summary>The site's real gRPC listener, when wire probes are attached.</summary>
|
||||
public SiteWireStreamHost? WireHost => _wireHost;
|
||||
|
||||
/// <summary>Instance Actors created on this site.</summary>
|
||||
public IReadOnlyList<IActorRef> InstanceActors => _instanceActors;
|
||||
|
||||
@@ -344,12 +355,63 @@ public sealed class SiteRuntimeFixture : IAsyncDisposable
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Starts the site's real Kestrel/gRPC listener and attaches
|
||||
/// <paramref name="count"/> subscribers through the production
|
||||
/// <see cref="SiteStreamGrpcClient"/> — one channel per site with the streams
|
||||
/// multiplexed on it, exactly the production shape (central keys one client per
|
||||
/// <c>(siteId, endpoint)</c>). Instance selection uses the same stride as
|
||||
/// <see cref="AttachStreamProbes"/>, so the wire probes watch the SAME instances
|
||||
/// as the in-process probes and the two latency distributions are directly
|
||||
/// comparable within one run.
|
||||
/// </summary>
|
||||
/// <param name="count">Wire subscribers to attach.</param>
|
||||
/// <param name="latency">Histogram that receives end-to-end (over-the-wire) latencies.</param>
|
||||
/// <param name="log">Progress sink for fault reporting.</param>
|
||||
/// <returns>
|
||||
/// A task that completes when the listener is up and every stream has been
|
||||
/// initiated. Stream ESTABLISHMENT finishes asynchronously — comfortably inside
|
||||
/// the subscribe-settle + settle windows that separate attach from measurement.
|
||||
/// </returns>
|
||||
public async Task AttachWireStreamProbesAsync(int count, LatencyHistogram latency, Action<string> log)
|
||||
{
|
||||
_wireHost = await SiteWireStreamHost.StartAsync(this);
|
||||
_wireClient = new SiteStreamGrpcClient(_wireHost.Endpoint, NullLogger.Instance);
|
||||
_wireCts = new CancellationTokenSource();
|
||||
|
||||
var stride = Math.Max(1, _config.InstancesPerSite / Math.Max(1, count));
|
||||
for (var p = 0; p < count; p++)
|
||||
{
|
||||
var instanceIndex = Math.Min(p * stride, _config.InstancesPerSite - 1);
|
||||
var subscriber = WireStreamSubscriber.Attach(
|
||||
_wireClient, InstanceName(instanceIndex), $"wire-{SiteId}-p{p:D2}",
|
||||
latency, log, _wireCts.Token);
|
||||
_wireSubscribers.Add(subscriber);
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>Total tag paths registered across this site's connections.</summary>
|
||||
public int TotalTagPaths => TagPathsByConnection.Sum(list => list.Count);
|
||||
|
||||
/// <inheritdoc />
|
||||
public async ValueTask DisposeAsync()
|
||||
{
|
||||
// Wire path first: cancel the client streams, wait for their loops to exit,
|
||||
// then drop the channel and the listener — the same order production teardown
|
||||
// takes (client goes away, server cancels, actor system last).
|
||||
if (_wireCts is not null)
|
||||
{
|
||||
await _wireCts.CancelAsync();
|
||||
foreach (var subscriber in _wireSubscribers)
|
||||
await subscriber.WaitForShutdownAsync();
|
||||
_wireCts.Dispose();
|
||||
}
|
||||
|
||||
if (_wireClient is not null)
|
||||
await _wireClient.DisposeAsync();
|
||||
if (_wireHost is not null)
|
||||
await _wireHost.DisposeAsync();
|
||||
|
||||
foreach (var probe in _probes)
|
||||
await probe.DisposeAsync();
|
||||
|
||||
|
||||
@@ -27,6 +27,14 @@
|
||||
<ConcurrentGarbageCollection>true</ConcurrentGarbageCollection>
|
||||
</PropertyGroup>
|
||||
|
||||
<ItemGroup>
|
||||
<!-- Wire-path probes: each site hosts the production SiteStreamGrpcServer on a
|
||||
real Kestrel h2c listener so the R2 batcher + serialization + HTTP/2 flow
|
||||
control sit inside the measured path (see Probes/SiteWireStreamHost.cs). -->
|
||||
<FrameworkReference Include="Microsoft.AspNetCore.App" />
|
||||
<PackageReference Include="Grpc.AspNetCore" />
|
||||
</ItemGroup>
|
||||
|
||||
<ItemGroup>
|
||||
<ProjectReference Include="../../src/ZB.MOM.WW.ScadaBridge.Commons/ZB.MOM.WW.ScadaBridge.Commons.csproj" />
|
||||
<ProjectReference Include="../../src/ZB.MOM.WW.ScadaBridge.SiteRuntime/ZB.MOM.WW.ScadaBridge.SiteRuntime.csproj" />
|
||||
|
||||
@@ -270,6 +270,29 @@ public class SecuredWriteHandlerTests : TestKit, IDisposable
|
||||
return row;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Arms the reject compare-and-swap to succeed (this verifier wins the
|
||||
/// Pending->Rejected transition). NSubstitute's unarmed default for the
|
||||
/// bool-returning CAS is <c>false</c>, i.e. "lost the race", so every reject test
|
||||
/// that expects the decision to land must arm it explicitly (arch-review
|
||||
/// 2026-08-19 P6-1).
|
||||
/// </summary>
|
||||
private void ArmRejectCasSuccess(long id) =>
|
||||
_securedWriteRepo.TryMarkRejectedAsync(
|
||||
id, Arg.Any<string>(), Arg.Any<string?>(), Arg.Any<DateTime>(), Arg.Any<CancellationToken>())
|
||||
.Returns(true);
|
||||
|
||||
/// <summary>
|
||||
/// Lets the best-effort audit path settle, then asserts NOTHING was written. Audit
|
||||
/// emission is awaited inline before the handler replies, so the reply itself is the
|
||||
/// happens-before edge; the short sleep is belt-and-braces against a late writer.
|
||||
/// </summary>
|
||||
private static void AssertNoAuditRows(List<AuditEvent> captured)
|
||||
{
|
||||
Thread.Sleep(250);
|
||||
Assert.Empty(captured);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Reject_BySubmittingUser_ReturnsError_StatusUnchanged()
|
||||
{
|
||||
@@ -284,17 +307,15 @@ public class SecuredWriteHandlerTests : TestKit, IDisposable
|
||||
var response = ExpectMsg<ManagementError>(TimeSpan.FromSeconds(5));
|
||||
Assert.Equal("COMMAND_FAILED", response.ErrorCode);
|
||||
_securedWriteRepo.DidNotReceiveWithAnyArgs().UpdateAsync(default!, default);
|
||||
_securedWriteRepo.DidNotReceiveWithAnyArgs().TryMarkRejectedAsync(
|
||||
default, default!, default, default, default);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Reject_ByDifferentUser_FlipsStatusToRejected()
|
||||
public void Reject_ByDifferentUser_CasFlipsStatusToRejected_NoBlindUpdate()
|
||||
{
|
||||
SeedPendingWrite(7, operatorUser: "alice");
|
||||
|
||||
PendingSecuredWrite? updated = null;
|
||||
_securedWriteRepo
|
||||
.When(r => r.UpdateAsync(Arg.Any<PendingSecuredWrite>(), Arg.Any<CancellationToken>()))
|
||||
.Do(ci => updated = ci.Arg<PendingSecuredWrite>());
|
||||
ArmRejectCasSuccess(7);
|
||||
|
||||
var actor = CreateActor();
|
||||
var envelope = Envelope(new RejectSecuredWriteCommand(7, "not authorized"), "bob", "Verifier");
|
||||
@@ -303,13 +324,21 @@ public class SecuredWriteHandlerTests : TestKit, IDisposable
|
||||
|
||||
var response = ExpectMsg<ManagementSuccess>(TimeSpan.FromSeconds(5));
|
||||
Assert.Equal(envelope.CorrelationId, response.CorrelationId);
|
||||
Assert.NotNull(updated);
|
||||
Assert.Equal("Rejected", updated!.Status);
|
||||
Assert.Equal("bob", updated.VerifierUser);
|
||||
Assert.Equal("not authorized", updated.VerifierComment);
|
||||
Assert.NotNull(updated.DecidedAtUtc);
|
||||
// Operator fields are preserved (fully-populated update).
|
||||
Assert.Equal("alice", updated.OperatorUser);
|
||||
|
||||
// The decision goes through the conditional Pending->Rejected CAS, exactly once,
|
||||
// carrying the verifier identity + comment (arch-review 2026-08-19 P6-1).
|
||||
_securedWriteRepo.Received(1).TryMarkRejectedAsync(
|
||||
7, "bob", "not authorized", Arg.Any<DateTime>(), Arg.Any<CancellationToken>());
|
||||
// The blind full-row overwrite is gone — it could land mid-relay on a row a
|
||||
// concurrent approve already owned.
|
||||
_securedWriteRepo.DidNotReceiveWithAnyArgs().UpdateAsync(default!, default);
|
||||
|
||||
// The projected row reflects the committed decision.
|
||||
Assert.Contains("Rejected", response.JsonData);
|
||||
Assert.Contains("bob", response.JsonData);
|
||||
Assert.Contains("not authorized", response.JsonData);
|
||||
// Operator fields are preserved.
|
||||
Assert.Contains("alice", response.JsonData);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
@@ -317,6 +346,7 @@ public class SecuredWriteHandlerTests : TestKit, IDisposable
|
||||
{
|
||||
var row = SeedPendingWrite(7, operatorUser: "alice");
|
||||
row.Status = "Approved";
|
||||
var captured = CaptureAuditEvents();
|
||||
|
||||
var actor = CreateActor();
|
||||
var envelope = Envelope(new RejectSecuredWriteCommand(7, null), "bob", "Verifier");
|
||||
@@ -326,6 +356,79 @@ public class SecuredWriteHandlerTests : TestKit, IDisposable
|
||||
var response = ExpectMsg<ManagementError>(TimeSpan.FromSeconds(5));
|
||||
Assert.Equal("COMMAND_FAILED", response.ErrorCode);
|
||||
_securedWriteRepo.DidNotReceiveWithAnyArgs().UpdateAsync(default!, default);
|
||||
_securedWriteRepo.DidNotReceiveWithAnyArgs().TryMarkRejectedAsync(
|
||||
default, default!, default, default, default);
|
||||
// Nothing was decided, so no Reject audit row may exist.
|
||||
AssertNoAuditRows(captured);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Reject_WhenCasLosesRace_ReturnsConflict_WritesNothing_NoRejectAudit()
|
||||
{
|
||||
// The P6-1 race, from the reject side: the status read at the top of the handler
|
||||
// saw Pending, but a concurrent approve won the Pending->Approved CAS and entered
|
||||
// its multi-second WriteTagAsync relay before this reject reached the store. The
|
||||
// reject must lose (CAS returns false), record NOTHING — no status write, no
|
||||
// audit row — and report the status that actually won, so the two-person-control
|
||||
// record can never say Rejected for a write that physically executed.
|
||||
SeedPendingWrite(7, operatorUser: "alice");
|
||||
_securedWriteRepo.TryMarkRejectedAsync(
|
||||
7, Arg.Any<string>(), Arg.Any<string?>(), Arg.Any<DateTime>(), Arg.Any<CancellationToken>())
|
||||
.Returns(false);
|
||||
// First read (the handler's stale-read guard) sees Pending; the post-CAS re-read
|
||||
// observes the status the concurrent approve committed.
|
||||
var stalePending = new PendingSecuredWrite
|
||||
{
|
||||
Id = 7, SiteId = "SITE1", ConnectionName = "Mx1", TagPath = "Tag.A",
|
||||
ValueJson = "true", ValueType = "Boolean", Status = "Pending",
|
||||
OperatorUser = "alice", SubmittedAtUtc = DateTime.UtcNow
|
||||
};
|
||||
var approvedByRival = new PendingSecuredWrite
|
||||
{
|
||||
Id = 7, SiteId = "SITE1", ConnectionName = "Mx1", TagPath = "Tag.A",
|
||||
ValueJson = "true", ValueType = "Boolean", Status = "Approved",
|
||||
OperatorUser = "alice", VerifierUser = "carol",
|
||||
SubmittedAtUtc = DateTime.UtcNow, DecidedAtUtc = DateTime.UtcNow
|
||||
};
|
||||
_securedWriteRepo.GetAsync(7, Arg.Any<CancellationToken>())
|
||||
.Returns(stalePending, approvedByRival);
|
||||
var captured = CaptureAuditEvents();
|
||||
|
||||
var actor = CreateActor();
|
||||
actor.Tell(Envelope(new RejectSecuredWriteCommand(7, "no"), "bob", "Verifier"));
|
||||
|
||||
var response = ExpectMsg<ManagementError>(TimeSpan.FromSeconds(5));
|
||||
Assert.Equal("COMMAND_FAILED", response.ErrorCode);
|
||||
// The conflict names the status that won the race.
|
||||
Assert.Contains("Approved", response.Error);
|
||||
Assert.Contains("cannot be rejected", response.Error);
|
||||
|
||||
_securedWriteRepo.Received(1).TryMarkRejectedAsync(
|
||||
7, "bob", "no", Arg.Any<DateTime>(), Arg.Any<CancellationToken>());
|
||||
_securedWriteRepo.DidNotReceiveWithAnyArgs().UpdateAsync(default!, default);
|
||||
AssertNoAuditRows(captured);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Approve_OnAlreadyRejectedRow_ReturnsError_NoCas_NoRelay_NoAudit()
|
||||
{
|
||||
// The mirror case: a rejected row can never be approved/executed afterwards.
|
||||
var row = SeedPendingWrite(7, operatorUser: "alice");
|
||||
row.Status = "Rejected";
|
||||
row.VerifierUser = "carol";
|
||||
var captured = CaptureAuditEvents();
|
||||
|
||||
var actor = CreateActor();
|
||||
actor.Tell(Envelope(new ApproveSecuredWriteCommand(7, "ok"), "bob", "Verifier"));
|
||||
|
||||
var response = ExpectMsg<ManagementError>(TimeSpan.FromSeconds(5));
|
||||
Assert.Equal("COMMAND_FAILED", response.ErrorCode);
|
||||
Assert.Contains("Rejected", response.Error);
|
||||
_securedWriteRepo.DidNotReceiveWithAnyArgs().TryMarkApprovedAsync(
|
||||
default, default!, default, default, default);
|
||||
_securedWriteRepo.DidNotReceiveWithAnyArgs().UpdateAsync(default!, default);
|
||||
Assert.Equal(0, _comms.CallCount); // the device write is NEVER relayed
|
||||
AssertNoAuditRows(captured);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
@@ -503,6 +606,7 @@ public class SecuredWriteHandlerTests : TestKit, IDisposable
|
||||
_securedWriteRepo.TryMarkApprovedAsync(
|
||||
7, Arg.Any<string>(), Arg.Any<string?>(), Arg.Any<DateTime>(), Arg.Any<CancellationToken>())
|
||||
.Returns(false);
|
||||
var captured = CaptureAuditEvents();
|
||||
|
||||
var actor = CreateActor();
|
||||
var envelope = Envelope(new ApproveSecuredWriteCommand(7, "race"), "bob", "Verifier");
|
||||
@@ -513,6 +617,9 @@ public class SecuredWriteHandlerTests : TestKit, IDisposable
|
||||
Assert.Equal("COMMAND_FAILED", response.ErrorCode);
|
||||
Assert.Contains("already decided", response.Error);
|
||||
Assert.Equal(0, _comms.CallCount);
|
||||
_securedWriteRepo.DidNotReceiveWithAnyArgs().UpdateAsync(default!, default);
|
||||
// A lost approve decides nothing, so it may leave no Approve audit row behind.
|
||||
AssertNoAuditRows(captured);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
@@ -751,8 +858,10 @@ public class SecuredWriteHandlerTests : TestKit, IDisposable
|
||||
|
||||
var resp = ExpectMsg<ManagementError>(TimeSpan.FromSeconds(5));
|
||||
Assert.Contains("expired", resp.Error, StringComparison.OrdinalIgnoreCase);
|
||||
// The reject path never overwrites the row (the CAS expiry owns the transition).
|
||||
// The reject path never writes the row (the CAS expiry owns the transition).
|
||||
_securedWriteRepo.DidNotReceiveWithAnyArgs().UpdateAsync(default!, default);
|
||||
_securedWriteRepo.DidNotReceiveWithAnyArgs().TryMarkRejectedAsync(
|
||||
default, default!, default, default, default);
|
||||
_securedWriteRepo.Received(1).TryMarkExpiredAsync(1, Arg.Any<DateTime>(), Arg.Any<CancellationToken>());
|
||||
}
|
||||
|
||||
@@ -829,6 +938,7 @@ public class SecuredWriteHandlerTests : TestKit, IDisposable
|
||||
public void Reject_EmitsOneRejectAuditRow_WithVerifier()
|
||||
{
|
||||
SeedPendingWrite(7, operatorUser: "alice");
|
||||
ArmRejectCasSuccess(7);
|
||||
var captured = CaptureAuditEvents();
|
||||
|
||||
var actor = CreateActor();
|
||||
@@ -1003,6 +1113,8 @@ public class SecuredWriteHandlerTests : TestKit, IDisposable
|
||||
actor.Tell(ScopedEnvelope(new RejectSecuredWriteCommand(42, "no"), "bob", ["3"], "Verifier"));
|
||||
ExpectMsg<ManagementUnauthorized>(TimeSpan.FromSeconds(5));
|
||||
_securedWriteRepo.DidNotReceiveWithAnyArgs().UpdateAsync(default!, default);
|
||||
_securedWriteRepo.DidNotReceiveWithAnyArgs().TryMarkRejectedAsync(
|
||||
default, default!, default, default, default);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
|
||||
+13
@@ -49,6 +49,7 @@ public class TargetScaleHarnessSmokeTests
|
||||
DebugProbeInterval = TimeSpan.FromSeconds(2),
|
||||
SubscribeSettleDuration = TimeSpan.FromSeconds(5),
|
||||
StreamProbesPerSite = 3,
|
||||
WireProbesPerSite = 2,
|
||||
StoreAndForwardDrainMessages = 200,
|
||||
SlowSubscriberEvents = 2_000,
|
||||
ResultsPath = Path.Combine(Path.GetTempPath(), $"loadharness-smoke-{Guid.NewGuid():N}.json"),
|
||||
@@ -64,6 +65,18 @@ public class TargetScaleHarnessSmokeTests
|
||||
Assert.True(result.StreamProbeReceived > 0, "Live stream subscribers received nothing.");
|
||||
Assert.Equal(0, result.DriverSkippedNoCallback);
|
||||
|
||||
// The wire path (real Kestrel + real SiteStreamGrpcClient, R2 batching
|
||||
// negotiated) carried traffic end to end: events crossed a real HTTP/2
|
||||
// socket, the batch-size histogram observed real frames, and nothing faulted.
|
||||
Assert.NotNull(result.WireStream);
|
||||
Assert.True(result.WireStream!.EventsReceived > 0,
|
||||
"Wire subscribers received nothing — the Kestrel/gRPC hop did not carry traffic.");
|
||||
Assert.True(result.WireStream.Latency.Count > 0,
|
||||
"No wire latency samples — the over-the-wire measurement path is broken.");
|
||||
Assert.True(result.WireStream.Batches.Frames > 0,
|
||||
"The batch-size MeterListener observed no frames — the R2 histogram path is broken.");
|
||||
Assert.Equal(0, result.WireStream.StreamFaults);
|
||||
|
||||
// Observability held up.
|
||||
Assert.Equal(config.Sites, result.SitesTrackedByAggregator);
|
||||
Assert.True(result.DebugSnapshotsCompleted > 0, "No debug snapshot completed.");
|
||||
|
||||
Reference in New Issue
Block a user