Merge Phase 1B: gRPC ApplyAck path (CentralCommandGrpcService + GrpcApplyAckSender) — live gate PASS 7/7
v2-ci / build (push) Successful in 5m54s
v2-ci / unit-tests (push) Failing after 19m21s

Node→central ApplyAck over gRPC behind MeshTransport:Mode=Grpc: CentralCommandGrpcService
on the AdminUI listener, GrpcApplyAckSender with per-endpoint ack deadlines (Critical fix
e54711a2 proven live in the 4b hang drill), NodeCommunicationActor wiring, plus the
phases 2-4 plan update (targeted-test policy + Task 2.0 must-fix).

Claude-Session: https://claude.ai/code/session_014WNM4vjoVksyyBraTXSZE1
This commit is contained in:
Joseph Doherty
2026-08-06 09:28:57 -04:00
17 changed files with 2067 additions and 57 deletions
@@ -649,20 +649,387 @@ if its cutover has to be reversed.
--- ---
## Phase 1B — node→central ack plane — **PENDING** ## Phase 1B — node→central ack plane — **PASS**
Branch `feat/grpc-central-ack` @ `________` (rebased on merged 1A). Rig-only flip: Branch `feat/grpc-central-ack` @ `4a2a3d3a` (rebased on `master`, which carries the merged 1A).
site-a-1/-2 → `Grpc`; site-b stays `ClusterClient`; central back on `ClusterClient` mode Rig-only flip: site-a-1/-2 → `Grpc`; site-b stays `ClusterClient`; central stays `ClusterClient`
(its command direction) — acks are governed by the **node's** mode, so this isolates 1B. (its command direction) — acks are governed by the **node's** mode, so this isolates 1B.
| # | Check | Result | | # | Check | Result |
|---|---|---| |---|---|---|
| 1 | Deploy e2e: site-a acks arrive via `ReportApply` (central `CentralCommandService` log lines carry site-a node ids), `NodeDeploymentState` rows land, deployment seals | PENDING | | 1 | Deploy e2e: site-a acks arrive via `ReportApply`, `NodeDeploymentState` rows land, deployment seals | **PASS** — deployment `53BD2626…` `Status=2` (Sealed) with all six rows `Status=1` (Applied) inside ~50 ms; both site-a nodes logged `Posted the ApplyAck … to http://central-1:4055` and central-1 logged the matching `ApplyAck … from site-a-1:4053 / site-a-2:4053 … handed to the central communication actor`. **10/10 acks delivered across the whole run**, 0 drops on any node |
| 2 | Site-b acks still arrive via ClusterClient in the same deployment — per-pair coexistence proven | PENDING | | 2 | Site-b acks still arrive via ClusterClient in the SAME deployment — per-pair coexistence | **PASS** — in `53BD2626…` site-b-1/-2 rows are `Applied` while **0** `ReportApply` lines name a site-b node on either central node, and site-b-1/-2 each hold a live `Connected to [akka.tcp://otopcua@central-2:4053/system/receptionist]`; conversely site-a-1/-2 hold **0** receptionist associations and created **0** ClusterClients. Each pair's only possible ack route is the one its mode selects |
| 3 | Exactly **one** ack row per site-a node per dispatch (no double-count from endpoint failover) | PENDING | | 3 | Exactly **one** ack row per site-a node per dispatch (no double-count from endpoint failover) | **PASS** — across 10 dispatches, `acks=1` for every (site-a node, deployment) pair on the happy path and on the `docker stop` failover. The **one** exception is the 4b pause drill (`458b8629…`, `acks=2` per node) and it is the documented at-least-once case, not a defect: the frozen server had in fact received the ack, and **the coordinator absorbed it without inflating the count — it reported `2/6 acks landed`, not `4/6`**. `NodeDeploymentState` holds exactly one row per node per deployment throughout |
| 4 | Kill central-1 before the apply completes: site-a acks fail over to central-2's endpoint; deployment still seals | PENDING | | 4 | Kill central-1 (`docker stop`) before the apply: site-a acks fail over to central-2; restart, verify convergence | **PASS** — with central-1 stopped, both site-a nodes logged the Information failover line `… to http://central-2:4055 after 1 endpoint(s) failed (http://central-1:4055 => Unavailable)` and central-2 accepted both **within the same second**; the deployment honestly ended `Status=4` (TimedOut) with central-1 alone at `Status=0` — sealing 6/6 is structurally impossible while one of the six expected nodes is down. After `start central-1` the pair reconverged (`members 2, unreachable 0`) and the next dispatch sealed **6/6** |
| 5 | Unauthenticated / wrong-cluster-key / missing `x-otopcua-cluster` probes against central `:4055` `CentralCommandService/ReportApply` ⇒ `PermissionDenied` (missing-header rejection is central-side by design) | PENDING | | **4b** | **Hang drill** (`docker pause central-1`): the per-endpoint FRESH deadline is what lets central-2 be reached at all | **PASS** — with central-1 **paused** (hung, not refusing), both site-a nodes logged `… to http://central-2:4055 after 1 endpoint(s) failed (http://central-1:4055 => DeadlineExceeded)` and the ack landed on central-2. **Observed failover latency ≈ 10 s** (dispatch accepted 13:03:51 → ack at 13:04:01), i.e. exactly one `AckTimeoutSeconds`. A single hoisted deadline would have entered the central-2 attempt already expired and dropped the ack; this is the live proof of the fix in `667639e6` |
| 6 | ConfigServe (`DeploymentArtifactService`) on the same listener unaffected: fetches succeed, 0 auth failures | PENDING | | 5 | Auth probes against central `:4055` `CentralCommandService/ReportApply` | **PASS** — 7/7 as designed: no bearer ⇒ `PermissionDenied`; SITE-B's key claiming SITE-A ⇒ `PermissionDenied`; no `x-otopcua-cluster` ⇒ `PermissionDenied`; unknown cluster `SITE-Z` with a valid key ⇒ `PermissionDenied`; unauthenticated against central-**2** ⇒ `PermissionDenied`; correct key+header with a **fake** deployment id ⇒ `{}` (auth passed, coordinator ignored it — **0 rows written** for that id); malformed id ⇒ `InvalidArgument` (auth passed, payload refused). Positive control: SITE-B key + SITE-B header also reaches the decoder |
| 6 | ConfigServe (`DeploymentArtifactService`) on the same listener unaffected | **PASS** — all four site nodes **Fetched + verified** a 28 296-byte artifact from `http://central-1:4055` on a real revision change, and a direct authenticated `Fetch` returned artifact bytes; unauthenticated `Fetch` is refused by its **own** interceptor (`Unauthenticated` / "Config-serve authentication failed."), never by the mesh one — which is the method-path scoping holding. **1** config-serve denial in the whole run: the deliberate probe |
### Provenance
`docker compose -f docker-dev/docker-compose.yml up -d --build` from the worktree
`/Users/dohertj2/Desktop/OtOpcUa/.worktrees/grpc-1b`, **exit 0**; image `otopcua-host:dev`
rebuilt to `4105cf631003` at 12:51:47 UTC. **All six host nodes plus `migrator`,
`cluster-seed` and `traefik` were recreated** (both one-shots exit 0); `sql` and its volume
were kept. Nothing else touched the stack during the run, so it is not a
concurrent-rebuild-invalidated one. Four rig-only Serilog `MinimumLevel:Override` Debug scopes
were added on all six nodes for this run only — `ZB.MOM.WW.OtOpcUa.Runtime.Communication` (the
sender's `Posted the ApplyAck` line), `ZB.MOM.WW.OtOpcUa.AdminUI.Grpc` (central's accept line),
`ZB.MOM.WW.OtOpcUa.ControlPlane.Communication` and `Akka.Cluster.Tools.Client` — because the
rig's default Serilog level is Information and checks 1–3 would otherwise have had no signal.
Flip and overrides reverted after the gate, never committed.
```
$ for c in central-1 central-2 site-a-1 site-a-2 site-b-1 site-b-2; do … done
central-1 running restarts=0 mode=ClusterClient hostingFailed=0 resolveErr=2
central-2 running restarts=0 mode=ClusterClient hostingFailed=0 resolveErr=2
site-a-1 running restarts=0 mode=Grpc hostingFailed=0 resolveErr=0
site-a-2 running restarts=0 mode=Grpc hostingFailed=0 resolveErr=0
site-b-1 running restarts=0 mode=ClusterClient hostingFailed=0 resolveErr=0
site-b-2 running restarts=0 mode=ClusterClient hostingFailed=0 resolveErr=0
$ curl … /health/ready → Healthy, members 2, unreachable 0 (MAIN)
$ … http://<site>:8080/health/ready → 200 on all four site nodes
```
### Checks 1 + 2 evidence — the two transports, side by side in one deployment
```
$ curl -X POST http://localhost:9200/api/deployments -H 'X-Api-Key: docker-dev-deploy-key' \
-d '{"createdBy":"1b-gate-check1"}'
{"outcome":"Accepted","deploymentId":"53bd2626-ce9f-4255-a964-24e4cf3c7ccb", …} HTTP 202
$ … SELECT Status FROM dbo.Deployment WHERE DeploymentId='53bd2626-…'
2 -- Sealed
$ … SELECT NodeId, Status, AppliedAtUtc FROM dbo.NodeDeploymentState WHERE DeploymentId='53bd2626-…'
central-1:4053|1|2026-08-06 12:55:53.909 site-a-1:4053|1|2026-08-06 12:55:53.930
central-2:4053|1|2026-08-06 12:55:53.890 site-a-2:4053|1|2026-08-06 12:55:53.936
site-b-1:4053|1|2026-08-06 12:55:53.915 site-b-2:4053|1|2026-08-06 12:55:53.924
# node side — the gRPC pair:
site-a-1 [12:55:53 DBG] Posted the ApplyAck for 53bd2626… (Applied) to http://central-1:4055.
site-a-2 [12:55:53 DBG] Posted the ApplyAck for 53bd2626… (Applied) to http://central-1:4055.
# central side — same two acks, named:
central-1 [12:55:53 DBG] ApplyAck for 53bd2626… from site-a-1:4053 (Applied) handed to the central communication actor
central-1 [12:55:53 DBG] ApplyAck for 53bd2626… from site-a-2:4053 (Applied) handed to the central communication actor
central-2 (none)
# per-node ReportApply accept counts, BOTH central nodes, this deployment:
central-1 site-a-1=1 site-a-2=1 site-b-1=0 site-b-2=0 central-1=0 central-2=0
central-2 site-a-1=0 site-a-2=0 site-b-1=0 site-b-2=0 central-1=0 central-2=0
```
The site-b half is proven by exclusion plus its live association, which is the same shape 1A
used for the site pairs:
```
$ for c in site-b-1 site-b-2; do docker logs … | grep -c "Posted the ApplyAck"; done → 0, 0
site-b-1 [12:53:02 INF] Connected to [akka.tcp://otopcua@central-2:4053/system/receptionist]
site-b-2 [12:53:07 INF] Connected to [akka.tcp://otopcua@central-2:4053/system/receptionist]
$ for c in site-a-1 site-a-2; do docker logs … | grep -c "Connected to .*receptionist"; done → 0, 0
$ for c in site-a-1 site-a-2 site-b-1 site-b-2; do grep -c "Mesh ClusterClient for cluster"; done → 0,0,0,0
```
Site-b rows are `Applied` while nothing on either central node logged a `ReportApply` for them,
and site-a holds no receptionist association at all — so neither pair could have used the
other's transport. Coexistence is per pair, in one deployment, from one dispatch.
### Check 3 evidence — one ack per dispatch, and the duplicate that proved the absorber
Across all ten dispatches, counting `ReportApply` accepts on **both** central nodes:
```
site-a-1 dep=53bd2626… acks=1 site-a-2 dep=53bd2626… acks=1
site-a-1 dep=d0869c9e… acks=1 site-a-2 dep=d0869c9e… acks=1
site-a-1 dep=17507001… acks=1 site-a-2 dep=17507001… acks=1 # the stop-failover dispatch
site-a-1 dep=b1e837a2… acks=1 site-a-2 dep=b1e837a2… acks=1
site-a-1 dep=a394eec6… acks=1 site-a-2 dep=a394eec6… acks=1
site-a-1 dep=9bb520a4… acks=1 site-a-2 dep=9bb520a4… acks=1
site-a-1 dep=458b8629… acks=2 site-a-2 dep=458b8629… acks=2 # the 4b PAUSE dispatch
site-a-1 dep=fc2b7b64… acks=1 site-a-2 dep=fc2b7b64… acks=1
site-a-1 dep=41e4935d… acks=1 site-a-2 dep=41e4935d… acks=1
site-a-1 dep=4126636e… acks=1 site-a-2 dep=4126636e… acks=1
$ … SELECT NodeId, COUNT(*) FROM dbo.NodeDeploymentState WHERE DeploymentId='53bd2626-…' GROUP BY NodeId
central-1:4053|1 central-2:4053|1 site-a-1:4053|1 site-a-2:4053|1 site-b-1:4053|1 site-b-2:4053|1
```
A refused endpoint (`docker stop`) cannot produce a duplicate — the server never saw the call —
which is why `17507001…` is 1 per node despite failing over. The pause drill is the one shape
that can, and it did:
```
central-2 [13:04:01 DBG] ApplyAck for 458b8629… from site-a-1:4053 (Applied) handed … # failover target
central-2 [13:04:01 DBG] ApplyAck for 458b8629… from site-a-2:4053 (Applied) handed …
central-1 [13:04:33 DBG] ApplyAck for 458b8629… from site-a-1:4053 (Applied) handed … # at `docker unpause`
central-1 [13:04:33 DBG] ApplyAck for 458b8629… from site-a-2:4053 (Applied) handed …
central-1 [13:05:03 INF] Coordinator recovered in-flight deployment 458b8629… (2/6 acks landed)
```
`2/6`, not `4/6`. The frozen central had in fact received both acks and replayed them on
unpause, and `ConfigPublishCoordinator.HandleAck`'s NodeId-keyed absorption made the duplicate
free — the exact claim `GrpcApplyAckSender`'s "at-least-once walk" remark makes, observed
rather than argued.
### Check 4 evidence — refusal failover
```
$ docker compose stop central-1 # 12:57:11
$ curl -X POST … -d '{"createdBy":"1b-gate-check4-failover"}' # 12:57:30
site-a-1 [12:57:30 INF] Posted the ApplyAck for 17507001… (Applied) to http://central-2:4055
after 1 endpoint(s) failed (http://central-1:4055 => Unavailable).
site-a-2 [12:57:30 INF] …identical…
central-2 [12:57:30 DBG] ApplyAck for 17507001… from site-a-1:4053 (Applied) handed …
central-2 [12:57:30 DBG] ApplyAck for 17507001… from site-a-2:4053 (Applied) handed …
$ … SELECT NodeId, Status FROM dbo.NodeDeploymentState WHERE DeploymentId='17507001-…'
central-1:4053|0 central-2:4053|1 site-a-1:4053|1 site-a-2:4053|1
site-b-1:4053|1 site-b-2:4053|1
$ … SELECT Status FROM dbo.Deployment … → 4 (TimedOut) after the 2-minute deadline
$ docker compose start central-1 # 13:00:10
$ curl … /health/ready → Healthy, members 2, unreachable 0
$ curl -X POST … -d '{"createdBy":"1b-gate-check4-converge"}' → Status=2, all six Status=1
```
**One operational detail worth carrying.** After central-1 returned, site-a-**1**'s cached
`GrpcChannel` to `http://central-1:4055` stayed in transient failure for roughly two minutes
while site-a-**2** reached the same restarted endpoint immediately — three dispatches apart:
```
13:01:25 site-a-1 → central-2 after (central-1 => Unavailable) | site-a-2 → central-1 OK
13:01:48 site-a-1 → central-2 after (central-1 => Unavailable) | site-a-2 → central-1 OK
13:02:49 site-a-1 → central-1 OK | site-a-2 → central-1 OK
$ docker exec otopcua-dev-site-a-1-1 bash -c '</dev/tcp/central-1/4055 && echo OPEN' → OPEN
$ central-1 mesh denials in that window → 0 # the call never reached the interceptor
```
TCP was open and central-1 logged nothing, so this is the client channel's own reconnect
backoff, not a server or auth problem. **No ack was lost** — the ordered walk absorbed it every
time and said so in the log. The Phase 2 reading: a central restart shifts that node's ack load
onto the surviving endpoint for a minute or two afterwards, which a soak's failover-line count
will show as a burst rather than a fault.
### Check 4b evidence — the hang, and why the fresh deadline is the whole check
```
$ docker pause otopcua-dev-central-1-1 # 13:03:35
$ curl … /health/ready → self akka.tcp://otopcua@central-2:4053 # traefik dropped central-1
$ curl -X POST … -d '{"createdBy":"1b-gate-check4b-hang"}' # 13:03:51 (202)
site-a-1 [13:04:01 INF] Posted the ApplyAck for 458b8629… (Applied) to http://central-2:4055
after 1 endpoint(s) failed (http://central-1:4055 => DeadlineExceeded).
site-a-2 [13:04:01 INF] …identical…
central-2 [13:04:01 DBG] ApplyAck for 458b8629… from site-a-1:4053 (Applied) handed …
central-2 [13:04:01 DBG] ApplyAck for 458b8629… from site-a-2:4053 (Applied) handed …
$ … SELECT NodeId, Status, AppliedAtUtc … WHERE DeploymentId='458b8629-…'
site-a-1:4053|1|2026-08-06 13:04:01.837 site-a-2:4053|1|2026-08-06 13:04:01.774
central-1:4053|0|NULL central-2:4053|0|NULL site-b-1:4053|0|NULL site-b-2:4053|0|NULL
$ docker unpause otopcua-dev-central-1-1 # 13:04:33
```
`DeadlineExceeded` — not `Unavailable` — is the discriminator: the first endpoint burned a full
`AckTimeoutSeconds` (10 s, the default; `13:03:51 → 13:04:01`) and central-2 was **still**
dialled with a live deadline. Under the hoisted deadline this branch shipped before
`667639e6`, that second attempt would have started already expired and the ack would have been
dropped with a Warning. Observed failover latency: **≈10 s, one whole `AckTimeoutSeconds`**.
**Collateral the drill produced, recorded because it is the honest picture of a `pause`.**
Pausing a cluster member is a heavier disturbance than stopping one: central-2 marked central-1
UNREACHABLE at 13:03:46 and auto-downed it at 13:04:01, and after `unpause` the two central
nodes each held a one-member view until central-1 was restarted (13:06:36). In that window the
**two gRPC acks landed and none of the four ClusterClient acks did** — central-2 and both
site-b nodes all logged `received DispatchDeployment` at 13:03:51 yet produced no ack the
coordinator saw, and the deployment ended `TimedOut` at `2/6` naming exactly those four:
```
central-1 [13:05:51 WRN] Deployment 458b8629… timed out after 00:02:00 (2/6 acks landed).
No ack from: central-1:4053, central-2:4053, site-b-1:4053, site-b-2:4053
central-1 [13:06:36 WRN] DriverHost central-1:4053: found orphan Applying row for deployment
458b8629-…; replaying
central-1 [13:06:37 INF] DriverHost central-1:4053: applied deployment 458b8629… (children=10)
```
Stated at the strength the evidence supports: the four ClusterClient nodes received the
dispatch and none of their acks reached the coordinator, while both gRPC nodes' acks did. The
site nodes emit no apply-completion line, so whether each of the four actually finished
applying is not determinable from their logs — what is determinable is that the ack did not
arrive and the deadline said so honestly. The boot-replay backstop then repaired central-1.
Not a 1B finding either way (no gRPC ack was involved), but it is the Phase 2 auto-down
behaviour showing up early, and the split view needed a manual `restart central-1` to heal.
### Check 5 evidence — the auth matrix on the ack endpoint
`-import-path` onto the in-repo proto (the node serves no reflection). `PermissionDenied` means
the interceptor refused; anything past it (`{}` or `InvalidArgument`) means auth **passed** and
only the payload was graded:
```
$ P=src/Core/ZB.MOM.WW.OtOpcUa.Commons/Protos
$ G="docker run --rm --network otopcua-dev_default -v $PWD/$P:/protos:ro fullstorydev/grpcurl \
-plaintext -import-path /protos -proto mesh_transport.proto"
$ BODY='{"deploymentId":"ffffffffffffffffffffffffffffffff","nodeId":"site-a-1:4053",
"outcome":"APPLY_OUTCOME_APPLIED","correlationId":"eeeeeeee…"}'
# 5a no bearer, cluster header present -> central-1
Code: PermissionDenied Message: Central command authentication failed.
# 5b Bearer dev-mesh-psk-SITE-B + cluster SITE-A -> central-1
Code: PermissionDenied Message: Central command authentication failed.
# 5c Bearer dev-mesh-psk-SITE-A, NO cluster header-> central-1
Code: PermissionDenied Message: Central command authentication failed.
# 5d Bearer dev-mesh-psk-SITE-A + cluster SITE-A, FAKE deployment id
{}
# 5e same, MALFORMED deployment id "not-a-guid"
Code: InvalidArgument Message: undecodable ApplyAck
# 5f Bearer dev-mesh-psk-SITE-B + cluster SITE-B (positive control)
Code: InvalidArgument Message: undecodable ApplyAck
# 5g Bearer dev-mesh-psk-SITE-A + cluster SITE-Z (unknown cluster)
Code: PermissionDenied Message: Central command authentication failed.
# 5h no bearer -> central-2
Code: PermissionDenied Message: Central command authentication failed.
```
The accepted fake ack changed nothing — the coordinator ignored a deployment it is not tracking:
```
$ … SELECT COUNT(*) FROM dbo.NodeDeploymentState WHERE DeploymentId='ffffffff-…' → 0
$ … SELECT COUNT(*) FROM dbo.Deployment WHERE DeploymentId='ffffffff-…' → 0
central-1 [13:08:08 DBG] ApplyAck for ffffffff… from site-a-1:4053 (Applied) handed to the
central communication actor
```
Central's denial log separates the three operator-facing causes even though the wire status is
deliberately uniform:
```
[13:08:07 WRN] Rejected a central command call to /mesh_transport.v1.CentralCommandService/ReportApply
claiming cluster SITE-A: no bearer token presented.
[13:08:07 WRN] … claiming cluster SITE-A: the presented bearer token did not match that cluster's
pre-shared key, no key could be established for it, or it is not a cluster this node knows…
[13:08:07 WRN] Rejected a central command call to …/ReportApply: no x-otopcua-cluster header, so there
is no cluster whose pre-shared key the caller could be graded against.
[13:08:25 WRN] … claiming cluster SITE-Z: … or it is not a cluster this node knows…
```
5g is the **known-cluster gate answering live**: a valid SITE-A key presented for `SITE-Z` is
refused in O(1), which also confirms the gate had in fact been published — see observation (b).
### Check 6 evidence — ConfigServe on the same listener
A real node-driven fetch needed a real revision change, so `ServerCluster.Notes` for `MAIN` was
edited, deployed, and **restored** (the revision hash returned to its pre-gate
`642b9d6f79b9…`, verified by a further dispatch):
```
$ … UPDATE dbo.ServerCluster SET Notes = '<original> | 1b-gate-fetch-probe' WHERE ClusterId='MAIN'
$ curl -X POST … → revisionHash 20af950fb211… (was 642b9d6f79b9…) Status=2
site-a-1 [13:09:31 INF] Fetched + verified artifact for 41e4935d… from http://central-1:4055 (28296 bytes).
site-a-2 [13:09:31 INF] …same… site-b-1 [13:09:32 INF] …same… site-b-2 [13:09:32 INF] …same…
$ … UPDATE dbo.ServerCluster SET Notes = '<original>' …
$ curl -X POST … → revisionHash 642b9d6f79b9… Status=2 # restored, byte-identical
```
Both interceptors on the one listener, each answering for its own service:
```
# authenticated Fetch, config-serve key:
$ $G -H 'authorization: Bearer configserve-docker-dev-key' -d '{"deploymentId":"FC2B7B64-…"}' \
central-1:4055 deployment_artifact.v1.DeploymentArtifactService/Fetch
{ "data": "eyJDbHVzdGVycyI6W3siQ2x1c3RlcklkIjoiTUFJTiIsIk5hbWUiOiJNYWluIGNsdXN0ZXIiL…" }
# unauthenticated Fetch — refused by the CONFIG-SERVE interceptor, not the mesh one:
Code: Unauthenticated Message: Config-serve authentication failed.
```
`Unauthenticated` here versus `PermissionDenied` on the mesh service is the method-path scoping
in `CentralCommandAuthInterceptor` holding: the two services share one `AddGrpc` pipeline and
neither interceptor grades the other's calls.
### Observations carried forward from the reviews
**(a) The I-4 shape is not exercisable on this rig, and that is a property of the rig, not a
skipped check.** `MeshPskProvider.TryGetPsk` consults `MeshTransport:ClusterPsks` **before** the
secret store and caches only store-resolved values, and docker-dev supplies
`MeshTransport__ClusterPsks__{MAIN,SITE-A,SITE-B}` on both central nodes. The
`InvalidateAll()` a contact-refresh tick performs (`ContactRefreshSeconds` = 60) therefore drops
an always-empty cache, and an ack arriving just after a tick pays no store read. No latency
delta was measurable and none should be: every ack in the run landed inside the same second it
was posted, refresh tick or not. A real deployment resolving `OT-MESH-PSK-<clusterId>` from the
store is where I-4 becomes observable — Phase 2 on a store-backed rig, not here.
**(b) The `Failed to resolve` Error fired on 4 of 4 admin boots — and healed every time.** Worse
odds than 1A's 2 of 3, and both dependencies lost the race on every boot:
```
central-1 [12:53:00 ERR] Failed to resolve IMeshCommandDialer … / IMeshPskProvider … # boot 1
central-1 [13:00:29 ERR] …both again… # boot 2 (check 4 restart)
central-1 [13:06:36 ERR] …both again… # boot 3 (4b restart)
central-2 [12:53:02 ERR] …both… # its single boot
central-1 [12:53:00 ERR] No IMeshPskProvider was supplied to this actor; the known-cluster gate
will NEVER be published and cached per-cluster keys will NEVER be dropped…
```
The second line is the one that reads worst, and 1B disproves it directly rather than by
argument: **every site-a ack central-1 accepted required that provider**, including the ones
after its third boot (13:07:36, 13:09:31, 13:09:59), and probe 5g's `SITE-Z` refusal is only
possible if `SetKnownClusters` had in fact been published. The memoize heals it; the Error text
does not say so. This is the 1A finding-5 carry-forward, now with a second, independent live
confirmation and a stronger case for the wording fix — the message contains the word "NEVER"
twice about something that self-corrects within one deploy.
**(c) Zero mesh interceptor denials outside the deliberate probes.** Whole-run counts on every
node and every interceptor:
```
central-1 telemetry=0 configserve=1 localdbsync=0 meshCentralCmd=4 nodeCmd=0
central-2 telemetry=0 configserve=0 localdbsync=0 meshCentralCmd=1 nodeCmd=0
site-a-1 telemetry=0 configserve=0 localdbsync=0 meshCentralCmd=0 nodeCmd=0
site-a-2 telemetry=0 configserve=0 localdbsync=0 meshCentralCmd=0 nodeCmd=0
site-b-1 telemetry=0 configserve=0 localdbsync=0 meshCentralCmd=0 nodeCmd=0
site-b-2 telemetry=0 configserve=0 localdbsync=0 meshCentralCmd=0 nodeCmd=0
```
`meshCentralCmd` = 4 on central-1 (probes 5a/5b/5c/5g) + 1 on central-2 (5h) = exactly the five
deliberate negatives; `configserve` = 1 = the deliberate 6b. Nothing else was refused anywhere.
**(d) Pre-existing rig noise, so a Phase 2 error grep is not surprised by it.** Site nodes log
`Unavailable … Name or service not known` from the historian client — the rig points
`ServerHistorian__Endpoint` at `http://histgw-absent.invalid:5222` on purpose (45 occurrences on
site-a-1, 1 on site-b-1, 4 on central-1 in this run). Nothing to do with the ack plane.
### Rig state after the gate — **restored, six healthy, all-ClusterClient, on the branch image**
The flip and the four Serilog overrides were reverted (`git checkout docker-dev/docker-compose.yml`,
`grep -c "GATE-1B ONLY"` → 0) and the nodes recreated on the committed compose. The simultaneous
recreate cold-started both central nodes at once and they self-formed one-member views — the
known simultaneous-cold-start split-brain the (default-OFF) `Cluster:BootstrapGuard` addresses —
healed with a single `restart central-2`:
```
central-1 running restarts=0 mode=ClusterClient image=4105cf631003 hostingFailed=0
central-2 running restarts=0 mode=ClusterClient image=4105cf631003 hostingFailed=0
site-a-1 running restarts=0 mode=ClusterClient image=4105cf631003 hostingFailed=0
site-a-2 running restarts=0 mode=ClusterClient image=4105cf631003 hostingFailed=0
site-b-1 running restarts=0 mode=ClusterClient image=4105cf631003 hostingFailed=0
site-b-2 running restarts=0 mode=ClusterClient image=4105cf631003 hostingFailed=0
$ for h in central-1 central-2; do … /health/ready; done
central-1 Healthy members 2 unreachable 0 leader akka.tcp://otopcua@central-1:4053
central-2 Healthy members 2 unreachable 0 leader akka.tcp://otopcua@central-1:4053
$ … http://<site>:8080/health/ready → 200 ×4
$ curl -X POST … -d '{"createdBy":"1b-gate-rig-restore-verify"}'
CDDA1D80-…|2 central-1|1 central-2|1 site-a-1|1 site-a-2|1 site-b-1|1 site-b-2|1
$ for c in site-a-1 site-a-2 site-b-1 site-b-2; do grep -c "Posted the ApplyAck"; done → 0,0,0,0
$ for c in central-1 central-2; do grep -c "Mesh ClusterClient for cluster"; done → 3 per boot
```
Zero gRPC acks after the rollback and three ClusterClients per central boot: the dark switch
rolls back cleanly in the ack direction too, which is what Phase 2 depends on if its cutover has
to be reversed. Rig configuration is byte-identical to its pre-gate state (`ServerCluster.Notes`
restored; revision hash back to `642b9d6f79b9…`).
### Phase 1B defects
**None.** No product code was changed during or after this gate.
--- ---
@@ -712,7 +1079,9 @@ is re-checked in the next gate (the ScadaBridge proportionate-gate rule).
| Check 4 acknowledge half — the condition actually acknowledging | 1A run 1 | The rig's one `ScriptedAlarm` sits Normal and its owning central node was down | — | **CLOSED in 1A run 2** — acked from the real `/alerts` page, `ACKNOWLEDGED` row observed | | Check 4 acknowledge half — the condition actually acknowledging | 1A run 1 | The rig's one `ScriptedAlarm` sits Normal and its owning central node was down | — | **CLOSED in 1A run 2** — acked from the real `/alerts` page, `ACKNOWLEDGED` row observed |
| Check 8 for telemetry + config-serve | 1A run 1 | Central dials both planes and never booted, so neither carried a call — 0 denials was vacuous | — | **CLOSED in 1A run 2** — both planes carried authenticated traffic with 0 denials | | Check 8 for telemetry + config-serve | 1A run 1 | Central dials both planes and never booted, so neither carried a call — 0 denials was vacuous | — | **CLOSED in 1A run 2** — both planes carried authenticated traffic with 0 denials |
| Check 3 central half — `RestartDriver`/`ReconnectDriver` **originating at the AdminUI** | 1A runs 1 + 2 | **Structurally undrivable, not a rig problem:** `<DriverStatusPanel` has 0 usages repo-wide (the 5 name hits are comments), no rendered page carries a Restart/Reconnect control, and the only non-script-analysis HTTP POST is `/api/deployments`. Pre-existing on `master`, not caused by this program | Composition the controller accepted: run 1's live node-side leg for both commands + run 2's check 4, which drives the identical `AdminOperationsActor.Dispatch` → `DispatchOverGrpc` origin path from the AdminUI for another command type; per-type difference is `ToEnvelope`, covered by the T0.2 goldens | **CLOSED by composition.** The orphaned panel itself is out of this program's scope — file separately if the control is meant to exist | | Check 3 central half — `RestartDriver`/`ReconnectDriver` **originating at the AdminUI** | 1A runs 1 + 2 | **Structurally undrivable, not a rig problem:** `<DriverStatusPanel` has 0 usages repo-wide (the 5 name hits are comments), no rendered page carries a Restart/Reconnect control, and the only non-script-analysis HTTP POST is `/api/deployments`. Pre-existing on `master`, not caused by this program | Composition the controller accepted: run 1's live node-side leg for both commands + run 2's check 4, which drives the identical `AdminOperationsActor.Dispatch` → `DispatchOverGrpc` origin path from the AdminUI for another command type; per-type difference is `ToEnvelope`, covered by the T0.2 goldens | **CLOSED by composition.** The orphaned panel itself is out of this program's scope — file separately if the control is meant to exist |
| `Failed to resolve` Error text says "continuing without it", which reads permanent although first use retries; and it will trip a soak's error grep on a healthy node | 1A run 2 (finding 5) | Cosmetic/observability, not behaviour | The self-heal proven live (finding 5) | Phase 2 | | `Failed to resolve` Error text says "continuing without it", which reads permanent although first use retries; and it will trip a soak's error grep on a healthy node | 1A run 2 (finding 5) | Cosmetic/observability, not behaviour | The self-heal proven live (finding 5), and **re-confirmed independently in 1B**: 4 of 4 admin boots lost the race, yet every site-a ack central-1 accepted and probe 5g's `SITE-Z` refusal both require the provider it claimed would "NEVER" be published | Phase 2 |
| I-4 cold-PSK-cache latency after a contact-refresh tick | 1B | **Not exercisable on docker-dev, structurally:** `TryGetPsk` reads `MeshTransport:ClusterPsks` ahead of the store and caches only store-resolved values, and the rig configures that map — so `InvalidateAll()` clears an always-empty cache and no ack pays a store read | Every ack in the run landed inside the second it was posted, tick or no tick (1B observation (a)) | Phase 2, on a store-backed rig |
| A restarted central's cached `GrpcChannel` can stay in transient failure for ~2 min on the node that saw it die, while a sibling node reaches the same endpoint at once | 1B check 4 | Not a defect — the ordered walk absorbed it and logged the failover each time; no ack lost. TCP was open and central logged nothing, so it is the client channel's backoff | 3 dispatches across the window, 2 failed over and the 3rd went direct (1B check 4 evidence) | Phase 2 soak — expect a burst of failover lines after any central restart, not a fault |
### Defects the gates caught that the suite did not ### Defects the gates caught that the suite did not
@@ -24,8 +24,8 @@ posture).
|---|---|---|---|---|---| |---|---|---|---|---|---|
| 0 | Proto + mappers + goldens; both auth interceptors; `IMeshPskProvider`; options/validators (+`Grpc` value, SplitTopology relaxation); rig dark config; stale-comment fixes; FleetStatus issue filed | `feat/grpc-phase0-contract-auth` | **MERGED to `master`** @ `10b6d209` (2026-08-06) — all 6 tasks complete @ `23555ed0`; every task spec ✅ + code ✅ (Batch B on Opus with a closed fix loop); issues #542 + #543 filed | `10b6d209` | **PASS 4/4** | | 0 | Proto + mappers + goldens; both auth interceptors; `IMeshPskProvider`; options/validators (+`Grpc` value, SplitTopology relaxation); rig dark config; stale-comment fixes; FleetStatus issue filed | `feat/grpc-phase0-contract-auth` | **MERGED to `master`** @ `10b6d209` (2026-08-06) — all 6 tasks complete @ `23555ed0`; every task spec ✅ + code ✅ (Batch B on Opus with a closed fix loop); issues #542 + #543 filed | `10b6d209` | **PASS 4/4** |
| 1A | Central→node command plane (`NodeCommandService` on the Telemetry listener; `MeshGrpcCommandDialer`; third mode branch in `CentralCommunicationActor`) — **worktree A** | `feat/grpc-node-command` | **COMPLETE @ `e54883a8` — gate re-run PASSED 2026-08-06.** Run 1 @ `fbf7b708` FAILED on defect 4 (no admin node could boot, mode-independent); fixed by `c0475f2d` + `e54883a8` and re-gated. 1A.1 ✅ `64774652` + 1A.2 ✅ `6a862011` (+fixes `48dc2120`/`86633f14`) all reviewed spec ✅ + code ✅. **MERGED to `master`** @ `b360239e` (2026-08-06) | `b360239e` | **PASS 8/8** (check 3's AdminUI half closed by composition — the control is structurally unreachable, see carry-forwards) | | 1A | Central→node command plane (`NodeCommandService` on the Telemetry listener; `MeshGrpcCommandDialer`; third mode branch in `CentralCommunicationActor`) — **worktree A** | `feat/grpc-node-command` | **COMPLETE @ `e54883a8` — gate re-run PASSED 2026-08-06.** Run 1 @ `fbf7b708` FAILED on defect 4 (no admin node could boot, mode-independent); fixed by `c0475f2d` + `e54883a8` and re-gated. 1A.1 ✅ `64774652` + 1A.2 ✅ `6a862011` (+fixes `48dc2120`/`86633f14`) all reviewed spec ✅ + code ✅. **MERGED to `master`** @ `b360239e` (2026-08-06) | `b360239e` | **PASS 8/8** (check 3's AdminUI half closed by composition — the control is structurally unreachable, see carry-forwards) |
| 1B | Node→central ack plane (`CentralCommandService` on the ConfigServe listener; `GrpcApplyAckSender`; third branch in `NodeCommunicationActor`) — **worktree B**, rebases on 1A | `feat/grpc-central-ack` | **1B.1 ✅ `14dd07d2` + 1B.2 ✅ `a5a789d4` (+fixes `e54711a2`/`4d607575`)** — all reviews ✅ (1B.2 on Opus: 1 Critical found+fixed, see defect log); awaiting 1A merge → rebase → 1B.3 gate | — | PENDING | | 1B | Node→central ack plane (`CentralCommandService` on the ConfigServe listener; `GrpcApplyAckSender`; third branch in `NodeCommunicationActor`) — **worktree B**, rebased on merged 1A | `feat/grpc-central-ack` | **DONE — AWAITING PR/MERGE (user-owned)** @ `a397cdad` (rebased on `a400b415`); 1B.1 ✅ + 1B.2 ✅ (Opus: deadline Critical found+fixed) + gate **PASS 7/7** incl. the 4b hang drill (~10 s failover = exactly one AckTimeoutSeconds — live proof of the fix); at-least-once duplicate absorbed idempotently exactly as traced | — | **PASS 7/7** |
| 2 | Baseline reproduction on pre-flip build, then full-rig cutover (`OTOPCUA_MESH_MODE` default → `Grpc`) + soak + kill drills | (rig commit) | NOT STARTED | — | PENDING | | 2 | **Opens with Task 2.0 (MUST FIX): honest PreStart-race logging** (both gates: fires ~every admin boot, two false Errors — soak error-grep impossible without it). Then baseline reproduction on pre-flip build, full-rig cutover (`OTOPCUA_MESH_MODE` default → `Grpc`) + soak + kill drills. **Targeted-test policy in force (plan Phase 2 header): per-task unit runs targeted; ONE full-suite run per phase at the DoD** | (rig commit; 2.0 on master) | NOT STARTED | — | PENDING |
| 3 | Delete ClusterClient (code/HOCON/config/tests); SplitTopology requires `Grpc`; grep-gates; docs sweep incl. scadaproj index | `feat/grpc-phase3-deletion` | NOT STARTED | — | — | | 3 | Delete ClusterClient (code/HOCON/config/tests); SplitTopology requires `Grpc`; grep-gates; docs sweep incl. scadaproj index | `feat/grpc-phase3-deletion` | NOT STARTED | — | — |
| 4 | Terminal live gate on the post-deletion image (8 checks, provenance-guarded) | (on `master`) | NOT STARTED | — | PENDING | | 4 | Terminal live gate on the post-deletion image (8 checks, provenance-guarded) | (on `master`) | NOT STARTED | — | PENDING |
@@ -145,6 +145,9 @@ Checks a gate could not drive, named + carried to the next gate (proportionate-g
| Register item 3 (DI role-scoping pin: admin resolves provider+dialer, driver-only does NOT) — container-activation test exists, role-scoping pin outstanding | 1A.2 | `It_activates_from_the_container_despite_the_optional_client_factory_parameter`; starter-order half now covered by `ControlPlaneStarterOrderTests` | Phase 2 (or 1B.3) — note defect 4 shows the missing pin is not merely role-scoping but **starter-order**: the container-activation test resolves from a collection without `AddZbSecrets` | | Register item 3 (DI role-scoping pin: admin resolves provider+dialer, driver-only does NOT) — container-activation test exists, role-scoping pin outstanding | 1A.2 | `It_activates_from_the_container_despite_the_optional_client_factory_parameter`; starter-order half now covered by `ControlPlaneStarterOrderTests` | Phase 2 (or 1B.3) — note defect 4 shows the missing pin is not merely role-scoping but **starter-order**: the container-activation test resolves from a collection without `AddZbSecrets` |
| `Resolve`-failure Error wording says "continuing without it" (reads permanent — the memoize heals it at first use) and the line fires on real boots (2/3 in the gate) — reword or whitelist it before the Phase 2 soak error-grep, else a healthy central scores as failing the soak | 1A.3 re-run | heal proven live: singleton moved onto the race-losing node, fan clean, no dialer-missing error | Phase 2 (soak grep design) | | `Resolve`-failure Error wording says "continuing without it" (reads permanent — the memoize heals it at first use) and the line fires on real boots (2/3 in the gate) — reword or whitelist it before the Phase 2 soak error-grep, else a healthy central scores as failing the soak | 1A.3 re-run | heal proven live: singleton moved onto the race-losing node, fan clean, no dialer-missing error | Phase 2 (soak grep design) |
| A repaired `GrpcPort` takes effect only at the next contact refresh (≤60 s): a deploy 5 s post-restore was still skipped and honestly TimedOut — by design, operator-visible; document if it surprises in Phase 2 | 1A.3 re-run | check 7 evidence | Phase 2 (observe) | | A repaired `GrpcPort` takes effect only at the next contact refresh (≤60 s): a deploy 5 s post-restore was still skipped and honestly TimedOut — by design, operator-visible; document if it surprises in Phase 2 | 1A.3 re-run | check 7 evidence | Phase 2 (observe) |
| I-4 (cold PSK cache on the inbound auth path after each refresh tick) is structurally unmeasurable on docker-dev — `ClusterPsks` config map answers ahead of the store, so the cache is always empty and `InvalidateAll` clears nothing | 1B.3 gate | every ack landed within the posting second | Phase 2+ (needs a store-backed rig) |
| After a central restart, a node's cached `GrpcChannel` can sit in transient failure ~2 min (client backoff; TCP open, sibling reaches the same endpoint) — no ack lost, the walk absorbs it loudly. Expect a failover-line burst after any central restart in the Phase 2 soak | 1B.3 gate | 3 dispatches: 2 failed over, 3rd direct | Phase 2 (soak expectations) |
| **The PreStart race fires on essentially EVERY admin boot** (4/4 in the 1B gate, 2/3 in 1A) and logs two FALSE Errors ("continuing without it", "gate will NEVER be published") that the memoize immediately heals — the wording fix is now a MUST before the Phase 2 soak error-grep, not a nicety | 1B.3 gate obs (b) | gate published + acks accepted on every one of those boots | Phase 2 opening task |
| `RevisionHash` moved on byte-identical config after a direct SQL `UPDATE` bumped `Script.RowVersion` (hash covers a column content-identity does not); determinism itself holds (3 no-change dispatches → identical hash) | 1A.3 re-run | incidental, out of program scope | note only | | `RevisionHash` moved on byte-identical config after a direct SQL `UPDATE` bumped `Script.RowVersion` (hash covers a column content-identity does not); determinism itself holds (3 no-change dispatches → identical hash) | 1A.3 re-run | incidental, out of program scope | note only |
| 1A checks 1/2/7 (deploy fan over gRPC, mixed-transport seal, NULL-`GrpcPort` skip) and the central half of 4 | 1A.3 gate run 1 | — | **CLOSED in 1A.3 run 2** — all driven live and PASS | | 1A checks 1/2/7 (deploy fan over gRPC, mixed-transport seal, NULL-`GrpcPort` skip) and the central half of 4 | 1A.3 gate run 1 | — | **CLOSED in 1A.3 run 2** — all driven live and PASS |
| Check 8 is vacuous for telemetry + config-serve — central never booted, so neither plane carried a call | 1A.3 gate run 1 | — | **CLOSED in 1A.3 run 2** — both planes carried authenticated traffic (28 KB artifact fetches from `:4055`, telemetry streams open on all six) with 0 denials | | Check 8 is vacuous for telemetry + config-serve — central never booted, so neither plane carried a call | 1A.3 gate run 1 | — | **CLOSED in 1A.3 run 2** — both planes carried authenticated traffic (28 KB artifact fetches from `:4055`, telemetry streams open on all six) with 0 denials |
+70 -6
View File
@@ -361,6 +361,58 @@ PASS; tracking updated. **STOP.**
## Phase 2 — Full-rig cutover + soak (sequential, after 1A+1B merged) ## Phase 2 — Full-rig cutover + soak (sequential, after 1A+1B merged)
> **Targeted-test policy (phases 2–4):** the full solution suite is large (~3,000+ tests,
> 40+ min wall). Per-task verification runs are TARGETED — run only the projects/filters the
> task touches (named per task below). ONE full-suite run happens per phase, at the phase DoD
> before the merge/STOP — never per task. The mesh-relevant targeted set, for reference:
>
> ```bash
> # mesh transport core (actor, dialer, provider, starter order)
> dotnet test tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/ --filter "FullyQualifiedName~Communication"
> # node-side actor + ack sender + registration wiring
> dotnet test tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/ --filter "FullyQualifiedName~Communication|FullyQualifiedName~ServiceCollectionExtensions"
> # interceptors + gRPC services (both sides)
> dotnet test tests/Server/ZB.MOM.WW.OtOpcUa.Host.Tests/
> # options/validators
> dotnet test tests/Core/ZB.MOM.WW.OtOpcUa.Cluster.Tests/ --filter "FullyQualifiedName~MeshTransport|FullyQualifiedName~SplitTopology"
> # proto contract + mapper goldens
> dotnet test tests/Core/ZB.MOM.WW.OtOpcUa.Commons.Tests/ --filter "FullyQualifiedName~MeshTransport"
> # central ack ingress
> dotnet test tests/Server/ZB.MOM.WW.OtOpcUa.AdminUI.Tests/ --filter "FullyQualifiedName~CentralCommandGrpcService"
> ```
### Task 2.0 (MUST FIX, opening task — before the flip and the soak): honest PreStart-race logging
**Classification:** small
**Estimated implement time:** ~15 min
**Parallelizable with:** none (first in phase)
**Files:**
- Modify: `src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Communication/CentralCommunicationActor.cs`
- Test: `tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/Communication/ControlPlaneStarterOrderTests.cs`, `CentralCommunicationActorTests.cs` (string/severity pins only)
Both live gates proved the `PreStart` resolve race fires on essentially EVERY admin boot
(1A: 2/3; 1B: 4/4) and logs two FALSE Errors — `Resolve`'s "continuing without it" (reads
permanent; the memoize heals it at first use) and the I-1 guard's "known-cluster gate will
NEVER be published" (it is published moments later — the 1B gate accepted acks and refused
`SITE-Z` on every such boot). A healthy central currently fails any error-grep, including
this phase's soak.
Required behavior after this task: a PreStart-time resolve failure logs at
**Debug/Info** ("not yet resolvable — starter ordering; will retry on first use"); the
**Error** fires only when a FIRST-USE retry also fails (the genuinely-permanent absence —
`Resolve` is already re-invoked by the memoizing properties, so this is a severity/wording
split by call site, not new machinery). The I-1 null-guard Errors in `PreStart` move to the
same first-use semantics or gain "if this persists past the first contact refresh" wording.
Net: **a healthy admin boot produces ZERO Warning/Error lines from this path** (pinned by a
test asserting the boot-time severity), while a truly-absent dependency still produces a
loud Error at first use (existing pins keep this). **Targeted tests only:**
`ControlPlaneStarterOrderTests` + `CentralCommunicationActorTests` filters. Small review
chain (code-reviewer). Commit `fix(mesh): PreStart resolve race logs honestly — quiet on
heal, loud on permanent absence`. Lands on `master` directly (post-1B-merge, pre-flip).
### The cutover
No feature branch; one commit flipping the rig default: No feature branch; one commit flipping the rig default:
`docker-dev/docker-compose.yml` `MeshTransport__Mode: "${OTOPCUA_MESH_MODE:-Grpc}"`. `docker-dev/docker-compose.yml` `MeshTransport__Mode: "${OTOPCUA_MESH_MODE:-Grpc}"`.
@@ -374,8 +426,11 @@ Then: full rebuild all-Grpc; run the Phase 2 check table (deploy e2e all six nod
Restart/Reconnect/AlarmCommand matrix; kill-central-1 ack failover; kill-site-node Restart/Reconnect/AlarmCommand matrix; kill-central-1 ack failover; kill-site-node
mid-dispatch = logged per-node failure + deadline backstop, no retry storm; PSK negatives mid-dispatch = logged per-node failure + deadline backstop, no retry storm; PSK negatives
against both services on running containers; **zero ClusterClient log lines under `Grpc`**; against both services on running containers; **zero ClusterClient log lines under `Grpc`**;
soak ≥ 2 h with 0 `PermissionDenied` noise — PSK drift detector). **Phase 2 DoD:** every soak ≥ 2 h with 0 `PermissionDenied` noise — PSK drift detector; the soak error-grep must be
check PASS in the live-gate doc; rig default committed; tracking updated. **STOP.** clean, which Task 2.0 makes possible on a healthy rig). Per-check verification is on the RIG;
unit runs during this phase are targeted (see policy above). **Phase 2 DoD:** every check
PASS in the live-gate doc; ONE full-suite run green on the cutover SHA; rig default
committed; tracking updated. **STOP.**
--- ---
@@ -407,6 +462,10 @@ client-settings asserts (keeps frame-size + DPS pins), `CentralCommunicationActo
`NodeCommunicationActorTests` / `MeshRouterDispatchTests` reshaped to Dps+Grpc. Grep-gates: `NodeCommunicationActorTests` / `MeshRouterDispatchTests` reshaped to Dps+Grpc. Grep-gates:
`rg -i "clusterclient|receptionist" src tests docker-dev docs` → only historical `rg -i "clusterclient|receptionist" src tests docker-dev docs` → only historical
plans/CHANGELOG; `rg "CentralContactPoints"` → empty. Commit sequence per step. plans/CHANGELOG; `rg "CentralContactPoints"` → empty. Commit sequence per step.
**Targeted tests per step, NOT the full suite:** steps 1–3 → the ControlPlane + Runtime
`Communication`/`ServiceCollectionExtensions` filters; step 4 → `MeshTransportHoconTests`;
steps 5–6 → the Cluster.Tests `MeshTransport|SplitTopology` filter + Host.Tests; step 7 →
the replaced/reshaped fixtures by name. The FULL suite runs once, at the Phase 3 DoD.
### Task 3.2: docs sweep ### Task 3.2: docs sweep
@@ -418,7 +477,9 @@ plans/CHANGELOG; `rg "CentralContactPoints"` → empty. Commit sequence per step
transport table, `docs/Telemetry.md` cross-refs, `OtOpcUa/CLAUDE.md` transport decisions, transport table, `docs/Telemetry.md` cross-refs, `OtOpcUa/CLAUDE.md` transport decisions,
`deferment.md` if applicable — and the **scadaproj index entry** (`~/Desktop/scadaproj/CLAUDE.md`, `deferment.md` if applicable — and the **scadaproj index entry** (`~/Desktop/scadaproj/CLAUDE.md`,
committed on scadaproj `main` and pushed, per its index-edit rule). Update tracking doc. committed on scadaproj `main` and pushed, per its index-edit rule). Update tracking doc.
**Phase 3 DoD:** suite green; grep-gates clean; docs merged. **STOP.** Task 3.2 itself is docs-only — no test runs beyond a build.
**Phase 3 DoD:** ONE full-suite run green on the branch tip (the phase's only full run);
grep-gates clean; docs merged. **STOP.**
--- ---
@@ -431,9 +492,12 @@ matrix all six nodes; (3) ack matrix + seal; (4) central-kill ack failover + sit
dispatch honesty; (5) mid-deploy central kill → deadline/replay backstops hold, no dupes; dispatch honesty; (5) mid-deploy central kill → deadline/replay backstops hold, no dupes;
(6) **boundary check: no Akka association between any site container and central** (was the (6) **boundary check: no Akka association between any site container and central** (was the
baseline gap); (7) restart discipline: full-rig restart clean, **zero baseline gap); (7) restart discipline: full-rig restart clean, **zero
receptionist/ClusterClient log lines anywhere**; (8) `MeshGrpcBoundaryTests` + full suite receptionist/ClusterClient log lines anywhere**; (8) `MeshGrpcBoundaryTests` (targeted) +
green on the merge SHA. Close any carry-forwards named by earlier gates. Tracking doc final the program's TERMINAL full-suite run, green on the merge SHA — this is Phase 4's one full
row; program CLOSED. run and the program's last; everything before it in this phase is rig checks + targeted
fixtures. Close any carry-forwards named by earlier gates (incl. the I-4 cold-cache
measurement if a store-backed rig exists by then — else re-carry with rationale). Tracking
doc final row; program CLOSED.
--- ---
@@ -149,10 +149,12 @@
"parallelizableWith": [ "parallelizableWith": [
"1A.1 (other worktree)" "1A.1 (other worktree)"
], ],
"status": "pending", "status": "completed",
"blockedBy": [ "blockedBy": [
5 5
] ],
"commit": "14dd07d2 (rebased 7e7df337)",
"review": "spec ✅ + code ✅"
}, },
{ {
"id": 10, "id": 10,
@@ -163,10 +165,12 @@
"parallelizableWith": [ "parallelizableWith": [
"1A.2 (other worktree)" "1A.2 (other worktree)"
], ],
"status": "pending", "status": "completed",
"blockedBy": [ "blockedBy": [
9 9
] ],
"commit": "a5a789d4 + fixes e54711a2/4d607575 (rebased 308c6aaa/667639e6/07bc1878)",
"review": "spec ✅ + code ✅ (opus, serial; 1 Critical fixed: per-endpoint deadline; fix loop closed)"
}, },
{ {
"id": 11, "id": 11,
@@ -175,11 +179,13 @@
"branch": "feat/grpc-central-ack", "branch": "feat/grpc-central-ack",
"classification": "small", "classification": "small",
"parallelizableWith": [], "parallelizableWith": [],
"status": "pending", "status": "completed",
"blockedBy": [ "blockedBy": [
8, 8,
10 10
] ],
"commit": "gate record a397cdad (rebase fixup 4a2a3d3a)",
"gate": "PASS 7/7 incl. 4b hang drill (~10s failover, per-endpoint deadline proven live)"
}, },
{ {
"id": 12, "id": 12,
@@ -191,8 +197,10 @@
"status": "pending", "status": "pending",
"blockedBy": [ "blockedBy": [
8, 8,
11 11,
] 16
],
"note": "Targeted-test policy applies (see plan Phase 2 header): per-task runs targeted; ONE full-suite run at the phase DoD."
}, },
{ {
"id": 13, "id": 13,
@@ -229,6 +237,20 @@
"blockedBy": [ "blockedBy": [
14 14
] ]
},
{
"id": 16,
"subject": "Task 2.0 (MUST FIX): PreStart-race logging honest — quiet on heal, loud on permanent absence",
"phase": "2",
"branch": "(master, post-1B-merge, pre-flip)",
"classification": "small",
"parallelizableWith": [],
"status": "pending",
"blockedBy": [
8,
11
],
"note": "Both gates: race fires ~every admin boot, two FALSE Errors; soak error-grep impossible without this. Targeted tests only (ControlPlaneStarterOrderTests + CentralCommunicationActorTests)."
} }
], ],
"lastUpdated": "2026-08-06" "lastUpdated": "2026-08-06"
@@ -94,9 +94,10 @@ public sealed class MeshTransportOptionsValidator : IValidateOptions<MeshTranspo
} }
} }
// The Grpc surface below is node-side: the PSK a driver presents (and requires), the central // The first Grpc block below is node-side: the PSK a driver presents (and requires), the
// endpoints it posts its ack to, and the two call deadlines. Scoped to the driver role for the // central endpoints it posts its ack to, and the two call deadlines (the admin-side rule that
// same reason the ClusterClient block above is scoped to its mode — an admin-only central node // follows it is a separate scope). Scoped to the driver role for the same reason the
// ClusterClient block above is scoped to its mode — an admin-only central node
// dials nodes rather than being dialled and resolves per-cluster keys at runtime through // dials nodes rather than being dialled and resolves per-cluster keys at runtime through
// MeshTransportOptions.ClusterPsks / the Secrets store, so requiring a single GrpcPsk and an // MeshTransportOptions.ClusterPsks / the Secrets store, so requiring a single GrpcPsk and an
// endpoint list of it would demand configuration it never reads. Roles come from // endpoint list of it would demand configuration it never reads. Roles come from
@@ -107,6 +108,7 @@ public sealed class MeshTransportOptionsValidator : IValidateOptions<MeshTranspo
var roles = RoleParser.ResolveEffectiveRoles( var roles = RoleParser.ResolveEffectiveRoles(
_configuration.GetSection("Cluster:Roles").Get<string[]>()); _configuration.GetSection("Cluster:Roles").Get<string[]>());
var isDriver = Array.IndexOf(roles, RoleParser.Driver) >= 0; var isDriver = Array.IndexOf(roles, RoleParser.Driver) >= 0;
var isAdmin = Array.IndexOf(roles, RoleParser.Admin) >= 0;
if (isGrpc && isDriver) if (isGrpc && isDriver)
{ {
@@ -166,6 +168,37 @@ public sealed class MeshTransportOptionsValidator : IValidateOptions<MeshTranspo
} }
} }
// The one admin-side obligation, and it is cross-read rather than a member of this options
// class: central hosts CentralCommandService on the SAME dedicated h2c listener as the
// config-serve endpoint (ConfigServe:GrpcListenPort) — a second cleartext HTTP/2 port would
// buy nothing but another firewall rule. With that port unset the service is never mapped, so
// every node's deployment ack has nowhere to land, and nothing errors: the deployment simply
// times out naming nodes that applied perfectly. Owned here rather than by ConfigServe's own
// validation because the requirement comes from MeshTransport:Mode, not from config-serve — a
// Dps or ClusterClient central node is legitimately portless.
if (isGrpc && isAdmin)
{
// Read as a string, the way SplitTopologyTransportValidator reads the modes:
// GetValue<int> THROWS on an unparseable value, and a validator whose whole job is to
// produce a readable failure must not be the thing that crashes on a typo. Missing,
// unparseable and non-positive all mean the same thing — nothing is bound — so they share
// one message.
var configServePort = _configuration[
$"{ConfigServeOptions.SectionName}:{nameof(ConfigServeOptions.GrpcListenPort)}"];
if (!int.TryParse(configServePort, out var port) || port <= 0)
{
errors.Add(
$"{ConfigServeOptions.SectionName}:"
+ $"{nameof(ConfigServeOptions.GrpcListenPort)} is not set to a positive port. "
+ $"Under {MeshTransportOptions.ModeGrpc} an admin node serves the central "
+ "command endpoint on that listener, and without it every node's deployment ack "
+ "has nowhere to land — the deploy would time out naming nodes that applied "
+ "perfectly. Supply it via the environment "
+ $"({ConfigServeOptions.SectionName}__"
+ $"{nameof(ConfigServeOptions.GrpcListenPort)}).");
}
}
return errors.Count == 0 return errors.Count == 0
? ValidateOptionsResult.Success ? ValidateOptionsResult.Success
: ValidateOptionsResult.Fail(string.Join(" ", errors)); : ValidateOptionsResult.Fail(string.Join(" ", errors));
@@ -0,0 +1,126 @@
using Akka.Actor;
using Akka.Hosting;
using Grpc.Core;
using Microsoft.Extensions.Logging;
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Deploy;
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Mesh;
using ZB.MOM.WW.OtOpcUa.Commons.Protos.MeshTransport.V1;
using ZB.MOM.WW.OtOpcUa.ControlPlane;
// The generated service container is itself named CentralCommandService; alias its nested server
// base so this impl can keep the natural name without colliding with the generated type.
using GeneratedServiceBase =
ZB.MOM.WW.OtOpcUa.Commons.Protos.MeshTransport.V1.CentralCommandService.CentralCommandServiceBase;
// Akka.Actor (needed for the single-argument Tell) also defines a Status. Every Status in this file
// is the gRPC one, so bind the name rather than qualifying each use.
using Status = Grpc.Core.Status;
namespace ZB.MOM.WW.OtOpcUa.AdminUI.Grpc;
/// <summary>
/// Central-side ingress for a node's deployment acknowledgment (ClusterClient→gRPC migration
/// Phase 1B). Decodes an <see cref="ApplyAckMessage"/> and hands it to the node-local
/// <c>CentralCommunicationActor</c> — the same actor an Akka ClusterClient receptionist delivered
/// to, whose forward-to-the-coordinator path is untouched. Registered only on admin-role nodes
/// and only when <c>ConfigServe:GrpcListenPort &gt; 0</c>.
/// </summary>
/// <remarks>
/// <para>
/// <b>The actor is resolved per call, not in the constructor.</b> Kestrel answers before
/// <c>Akka.Hosting</c> has finished spawning the actor graph, so a ctor lookup would throw
/// during the boot window and surface as an opaque <c>Unknown</c>. Resolving per call turns
/// that window into <see cref="StatusCode.Unavailable"/> — the retryable status. The node's
/// sender does not in fact retry this endpoint; it advances to the next central endpoint in
/// its list, where the ack lands on a node whose actor graph is up. Either way the deploy does
/// not record a permanent failure against a node that applied fine.
/// </para>
/// <para>
/// <b>Decoding is graded before the registry.</b> A message that cannot decode is
/// <see cref="StatusCode.InvalidArgument"/> even when the actor is also missing: Unavailable
/// is the status that says "try elsewhere", and moving a message that can never decode to
/// another endpoint only spends the second one too. The decode itself
/// never guesses (see <see cref="MeshTransportDtoMapper"/>) — an unspecified outcome, an
/// unparseable id or a blank node name is refused rather than resolved to a plausible value,
/// because the deploy path seals a revision on <c>Applied</c> and a guessed ack seals a
/// deployment nobody applied.
/// </para>
/// <para>
/// <b>The rejection log carries the wire correlation id verbatim.</b> The inner ids are
/// exactly what may have failed to parse, so the only value guaranteed readable at that point
/// is the raw string — which is what an operator has to correlate the rejection against the
/// node's own log.
/// </para>
/// <para>
/// <b>One-way semantics, preserved.</b> <c>Tell</c>, not <c>Ask</c>: the returned
/// <see cref="ReportAck"/> means "handed to the comm actor", never "the deployment is
/// complete" — the same promise the receptionist made. The comm actor forwards to the
/// coordinator singleton proxy, so an ack landing on either central node routes identically.
/// </para>
/// </remarks>
public sealed class CentralCommandGrpcService : GeneratedServiceBase
{
private readonly ActorRegistry _registry;
private readonly ILogger<CentralCommandGrpcService> _log;
/// <summary>Initializes a new instance of the <see cref="CentralCommandGrpcService"/> class.</summary>
/// <param name="registry">The actor registry holding the node's <c>CentralCommunicationActor</c>.</param>
/// <param name="log">Logger.</param>
public CentralCommandGrpcService(ActorRegistry registry, ILogger<CentralCommandGrpcService> log)
{
_registry = registry;
_log = log;
}
/// <inheritdoc />
public override Task<ReportAck> ReportApply(ApplyAckMessage request, ServerCallContext context)
{
ArgumentNullException.ThrowIfNull(request);
ArgumentNullException.ThrowIfNull(context);
ApplyAck ack;
try
{
// Exactly the three the mapper documents: an unspecified/unknown outcome
// (InvalidOperationException), an unparseable id (FormatException) and a blank node id
// (ArgumentException). Deliberately not a blanket catch — anything else escaping the
// mapper is a bug on this side, and reporting it to the caller as their InvalidArgument
// would send an operator to debug a node that sent a perfectly good ack.
ack = MeshTransportDtoMapper.FromProto(request);
}
catch (Exception ex) when (ex is InvalidOperationException or FormatException or ArgumentException)
{
// Logged with the RAW correlation id: the decode may have failed on the ids themselves.
_log.LogWarning(
ex,
"Rejected an ApplyAck (correlation '{CorrelationId}') from {Peer}: it could not be "
+ "decoded, so the deployment it names is NOT acknowledged on this node",
request.CorrelationId, context.Peer);
throw new RpcException(new Status(StatusCode.InvalidArgument, "undecodable ApplyAck"));
}
if (!_registry.TryGet<CentralCommunicationActorKey>(out var actor))
{
// Peer and correlation id are carried here for the same reason the decode-failure warning
// above carries them: these two rejections are the only ways an ack fails to reach the
// coordinator, and an operator has to correlate either against the node's own log. The
// decode has already succeeded on this path, so the DECODED correlation id is used —
// unlike above, where the raw wire string is the only value guaranteed readable.
_log.LogWarning(
"Received an ApplyAck for {DeploymentId} (correlation '{CorrelationId}') from {NodeId} "
+ "at {Peer} before the central communication actor was registered on this node; "
+ "refusing with Unavailable so the caller moves on to another central endpoint "
+ "rather than recording a failure against a node that applied",
ack.DeploymentId, ack.CorrelationId, ack.NodeId, context.Peer);
throw new RpcException(new Status(StatusCode.Unavailable, "central command actor not ready"));
}
// Tell, not Ask: the RPC's own response is the acknowledgment, and the comm actor's existing
// Forward-to-coordinator path is what it always was.
actor.Tell(ack);
_log.LogDebug(
"ApplyAck for {DeploymentId} from {NodeId} ({Outcome}) handed to the central communication actor",
ack.DeploymentId, ack.NodeId, ack.Outcome);
return Task.FromResult(new ReportAck());
}
}
@@ -744,6 +744,16 @@ if (hasDriver && syncListenPort > 0)
if (hasAdmin && configServeGrpcPort > 0) if (hasAdmin && configServeGrpcPort > 0)
{ {
app.MapGrpcService<DeploymentArtifactGrpcService>(); app.MapGrpcService<DeploymentArtifactGrpcService>();
// Central command ingress (ClusterClient→gRPC migration): the node→central deployment ack. It
// rides the SAME listener rather than claiming a port of its own — a second cleartext h2c port
// would buy nothing but another firewall rule, and both endpoints are admin-role and dialled by
// the same driver nodes. Deliberately NOT gated on MeshTransport:Mode: mapping it under Dps
// costs nothing (nothing dials it) whereas gating it would mean flipping the mode on a running
// fleet requires a central restart before any node can ack — the same reasoning that registers
// CentralCommunicationActor in both modes. CentralCommandAuthInterceptor (added at AddGrpc
// above) is the ONLY inbound auth and fail-closes per cluster.
app.MapGrpcService<CentralCommandGrpcService>();
} }
// Driver-node live-telemetry stream (per-cluster mesh Phase 5). Central dials in; gated by // Driver-node live-telemetry stream (per-cluster mesh Phase 5). Central dials in; gated by
@@ -0,0 +1,243 @@
using System.Collections.Concurrent;
using Grpc.Core;
using Grpc.Net.Client;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
using ZB.MOM.WW.OtOpcUa.Cluster;
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Deploy;
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Mesh;
using ZB.MOM.WW.OtOpcUa.Commons.Protos.MeshTransport.V1;
namespace ZB.MOM.WW.OtOpcUa.Runtime.Communication;
/// <summary>
/// Posts this node's <see cref="ApplyAck"/> to central's <c>CentralCommandService</c> as a unary
/// gRPC call (ClusterClient→gRPC migration Task 1B.2), the <c>MeshTransport:Mode = Grpc</c>
/// replacement for the ClusterClient <c>Send</c>.
/// </summary>
/// <remarks>
/// <para>
/// <b>One ordered pass, stopping at the first success.</b>
/// <see cref="MeshTransportOptions.CentralGrpcEndpoints"/> is an availability list, not a
/// fan-out list — the semantic of the <c>ClusterClient.Send</c> being replaced (<c>Send</c>,
/// deliberately not <c>SendToAll</c>).
/// </para>
/// <para>
/// <b>Delivery is at-most-once per SUCCESSFUL attempt, but the walk as a whole is
/// at-least-once</b>, and that is fine. An endpoint that processes the ack and then exceeds
/// its deadline before answering is indistinguishable here from one that never received it, so
/// the next endpoint is tried and central may see the ack twice.
/// <c>ConfigPublishCoordinator.HandleAck</c> absorbs that idempotently: acks land in a
/// <c>NodeId</c>-keyed dictionary behind an expected-set membership guard and a
/// stale-deployment guard, so a duplicate overwrites this node's own entry rather than
/// inflating the count, and the row upsert behind it is idempotent too. Duplicates therefore
/// cannot seal a deployment other nodes never applied — the single-endpoint rule is defence in
/// depth and a bound on wasted work, not a correctness dependency.
/// </para>
/// <para>
/// <b>No retry beyond that single pass, and no queue.</b> The receptionist buffered nothing
/// and retried nothing, and this is a like-for-like replacement, not an upgrade to delivery
/// guarantees. The deploy path's own apply deadline stays the only recovery mechanism; a
/// retry loop here would instead pile acks up behind an unreachable central and deliver a
/// burst of stale ones when it returned.
/// </para>
/// <para>
/// <b>All endpoints failing is a DROP — but a loud one.</b> That is the same outcome the
/// ClusterClient path had when it could not reach central; what changes is that the operator
/// now gets the node, the deployment, the correlation id and the status each endpoint
/// answered with, which is what distinguishes "central is down" from "central rejected this
/// node's key".
/// </para>
/// <para>
/// <b>The bearer never reaches a log.</b> It travels in the call headers on every attempt and
/// the failure log names endpoints and status codes only — a per-attempt dump of the
/// <see cref="Metadata"/> would write the cluster's pre-shared key into the node log of every
/// deployment that could not reach central.
/// </para>
/// <para>
/// <b>The cluster id comes from this node's own <c>cluster-&lt;id&gt;</c> Akka role</b>, read
/// through <see cref="RoleParser"/> off the same bound <c>Cluster:Roles</c> that
/// <c>ClusterRoleInfo</c> reads (the <c>OTOPCUA_ROLES</c> fallback is already applied to those
/// options by <c>AddOtOpcUaCluster</c>'s <c>PostConfigure</c>, so nothing is re-resolved
/// here). The two must agree: central grades the bearer against the key of the cluster named
/// in the header, so a node that authenticated as a different cluster than it gossips would be
/// rejected on every ack.
/// </para>
/// </remarks>
public sealed class GrpcApplyAckSender : IApplyAckSender, IDisposable
{
/// <summary>The cluster the caller claims to be; central grades the bearer against ITS key.</summary>
private const string ClusterHeader = "x-otopcua-cluster";
private readonly MeshTransportOptions _options;
private readonly string? _clusterId;
private readonly ILogger<GrpcApplyAckSender> _log;
private readonly Func<string, CentralCommandService.CentralCommandServiceClient> _clientFactory;
private readonly ConcurrentDictionary<string, GrpcChannel> _channels = new(StringComparer.Ordinal);
/// <summary>Initializes a new instance of the <see cref="GrpcApplyAckSender"/> class.</summary>
/// <param name="options">Mesh-transport options; supplies the endpoints, the PSK and the deadline.</param>
/// <param name="clusterOptions">
/// Cluster options, read only for this node's roles — the <c>cluster-&lt;id&gt;</c> one names
/// the cluster this node authenticates as.
/// </param>
/// <param name="log">Logger.</param>
/// <param name="clientFactory">
/// Seam for tests: builds the generated client for an endpoint. Defaults to a real
/// <see cref="GrpcChannel"/> (one cached per endpoint).
/// </param>
public GrpcApplyAckSender(
IOptions<MeshTransportOptions> options,
IOptions<AkkaClusterOptions> clusterOptions,
ILogger<GrpcApplyAckSender> log,
Func<string, CentralCommandService.CentralCommandServiceClient>? clientFactory = null)
{
ArgumentNullException.ThrowIfNull(options);
ArgumentNullException.ThrowIfNull(clusterOptions);
ArgumentNullException.ThrowIfNull(log);
_options = options.Value;
_log = log;
_clusterId = RoleParser.ClusterIdFromRoles(clusterOptions.Value.Roles);
_clientFactory = clientFactory ?? DefaultClientFactory;
if (_clusterId is null)
{
// Said once, at construction, rather than on every ack. Central cannot grade a bearer it
// has no cluster for, so every ack this node posts will come back PermissionDenied — and
// without this line that reads as a key mismatch rather than a missing role.
_log.LogWarning(
"This node carries no '{Prefix}<ClusterId>' role, so its deployment acks will be "
+ "posted without the {Header} header and central will reject every one of them. Add "
+ "the cluster-scoped role to Cluster:Roles (or OTOPCUA_ROLES).",
RoleParser.ClusterRolePrefix, ClusterHeader);
}
}
/// <inheritdoc />
public async Task SendAsync(ApplyAck ack, CancellationToken cancellationToken = default)
{
try
{
// INSIDE the try, not before it. This is an async method, so a guard clause above the try
// does not throw to the caller — it faults the returned task, which the caller discards.
// That would break this method's "never faults" promise in the one case a caller could
// never see.
ArgumentNullException.ThrowIfNull(ack);
await PostAsync(ack, cancellationToken);
}
catch (Exception ex)
{
// Deliberately unbounded, and deliberately not a rethrow. The caller discards this task
// (see IApplyAckSender), so anything escaping would be swallowed by the runtime and
// resurface as an unobserved exception at GC — a drop with no line in the node log, which
// is precisely the failure mode this transport replaces. Encoding faults and non-RpcException
// transport faults both land here.
//
// The ack is null-conditional because the guard above is one of the things this catches.
_log.LogWarning(
ex,
"Dropping the ApplyAck for {DeploymentId} (correlation {CorrelationId}) from {NodeId}: "
+ "posting it to central failed unexpectedly. It is not buffered and will not be "
+ "retried, so the deployment will time out at central naming this node even though it "
+ "applied successfully.",
ack?.DeploymentId, ack?.CorrelationId, ack?.NodeId);
}
}
private async Task PostAsync(ApplyAck ack, CancellationToken cancellationToken)
{
var endpoints = _options.CentralGrpcEndpoints;
if (endpoints.Length == 0)
{
_log.LogWarning(
"Dropping the ApplyAck for {DeploymentId} from {NodeId}: no "
+ "MeshTransport:CentralGrpcEndpoints are configured, so there is nowhere to post it. "
+ "The deployment will time out at central naming this node even though it applied.",
ack.DeploymentId, ack.NodeId);
return;
}
var message = MeshTransportDtoMapper.ToProto(ack);
// Hoisted: read-only, and the walk is strictly sequential, so one Metadata is shared safely
// across the attempts. The DEADLINE emphatically is not hoisted — see below.
var headers = new Metadata { { "authorization", $"Bearer {_options.GrpcPsk}" } };
if (_clusterId is not null)
headers.Add(ClusterHeader, _clusterId);
var failures = new List<string>(endpoints.Length);
foreach (var endpoint in endpoints)
{
try
{
// Computed FRESH per endpoint, which is what MeshTransport:AckTimeoutSeconds documents
// ("applied per endpoint, not per ack"). A single absolute deadline hoisted out of this
// loop defeats failover against the characteristic WAN failure: a central that HANGS
// (SYN black-hole, frozen host) rather than refusing. The first endpoint burns the whole
// timeout, and the second is then dialled with a deadline already in the past — it fails
// instantly as DeadlineExceeded without a packet leaving the node, so the surviving
// central is never actually tried.
var deadline = _options.AckTimeoutSeconds > 0
? DateTime.UtcNow.AddSeconds(_options.AckTimeoutSeconds)
: (DateTime?)null;
var client = _clientFactory(endpoint);
using var call = client.ReportApplyAsync(message, headers, deadline, cancellationToken);
await call;
if (failures.Count == 0)
{
_log.LogDebug(
"Posted the ApplyAck for {DeploymentId} ({Outcome}) to {Endpoint}.",
ack.DeploymentId, ack.Outcome, endpoint);
}
else
{
// Information, not Debug, because this is a FAILOVER: the ack landed, but only
// after one or more endpoints cost their full deadline. The deploy simply took
// that much longer, and without this line the only evidence is the absence of the
// all-endpoints-failed Warning — which reads identically to a healthy first-try.
_log.LogInformation(
"Posted the ApplyAck for {DeploymentId} ({Outcome}) to {Endpoint} after "
+ "{FailedCount} endpoint(s) failed ({Failures}).",
ack.DeploymentId, ack.Outcome, endpoint, failures.Count, string.Join("; ", failures));
}
// The single-endpoint rule, and it is this `return`: the walk stops at the first
// endpoint that accepted the ack and never goes back to one that did not.
return;
}
catch (RpcException ex)
{
// Status code only — never the exception's own message and never the headers, either
// of which could echo credential material back into the log.
failures.Add($"{endpoint} => {ex.StatusCode}");
}
}
_log.LogWarning(
"Dropping the ApplyAck for {DeploymentId} (correlation {CorrelationId}) from {NodeId}: all "
+ "{Count} central endpoint(s) refused it ({Failures}). It is not buffered and will not be "
+ "retried, so the deployment will time out at central naming this node even though it "
+ "applied successfully.",
ack.DeploymentId, ack.CorrelationId, ack.NodeId, endpoints.Length, string.Join("; ", failures));
}
private CentralCommandService.CentralCommandServiceClient DefaultClientFactory(string endpoint)
{
// h2c: GrpcChannel over an http:// address negotiates prior-knowledge HTTP/2. One channel per
// endpoint, cached and reused (creating a channel per ack would leak a socket per deployment).
var channel = _channels.GetOrAdd(endpoint, ep => GrpcChannel.ForAddress(ep));
return new CentralCommandService.CentralCommandServiceClient(channel);
}
/// <inheritdoc />
public void Dispose()
{
foreach (var channel in _channels.Values)
channel.Dispose();
_channels.Clear();
}
}
@@ -0,0 +1,40 @@
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Deploy;
namespace ZB.MOM.WW.OtOpcUa.Runtime.Communication;
/// <summary>
/// Posts this node's deployment acknowledgment to central over the <c>mesh_transport.v1</c>
/// command plane (ClusterClient→gRPC migration Phase 1B). The seam
/// <see cref="NodeCommunicationActor"/> uses under <c>MeshTransport:Mode = Grpc</c>, in place of
/// the <c>ClusterClient.Send</c> it uses under <c>ClusterClient</c>.
/// </summary>
/// <remarks>
/// <para>
/// <b>One ack per node per dispatch.</b> Whatever the implementation does about redundancy, it
/// posts to <i>one</i> central endpoint at a time rather than fanning out — the semantic of the
/// <c>ClusterClient.Send</c> this replaces (<c>Send</c>, deliberately not <c>SendToAll</c>).
/// This is <b>defence in depth, not a correctness dependency</b>: the coordinator counts acks
/// in a <c>NodeId</c>-keyed dictionary behind an expected-set membership guard
/// (<c>ConfigPublishCoordinator.HandleAck</c>), so a duplicate from one node overwrites its own
/// entry and cannot inflate the count or seal a deployment early. What fan-out would actually
/// cost is duplicated work at central, doubled deploy-ack rows written, and a transport whose
/// delivery count nobody can reason about — none of which is worth having.
/// </para>
/// <para>
/// <b>Fire-and-forget, and it never throws.</b> The caller is an actor that discards the
/// returned task, so an exception escaping here would become an unobserved task fault — the
/// silent drop this migration exists to remove. An implementation reports failure by logging
/// and returning, never by faulting the task.
/// </para>
/// </remarks>
public interface IApplyAckSender
{
/// <summary>Posts <paramref name="ack"/> to central, or logs why it could not be.</summary>
/// <param name="ack">The per-node apply acknowledgment.</param>
/// <param name="cancellationToken">Cancels the in-flight call.</param>
/// <returns>
/// A task that completes when the attempt finishes, successfully or not. It never faults —
/// see the remarks.
/// </returns>
Task SendAsync(ApplyAck ack, CancellationToken cancellationToken = default);
}
@@ -37,23 +37,31 @@ namespace ZB.MOM.WW.OtOpcUa.Runtime.Communication;
public sealed class NodeCommunicationActor : ReceiveActor public sealed class NodeCommunicationActor : ReceiveActor
{ {
private readonly IActorRef? _centralClient; private readonly IActorRef? _centralClient;
private readonly IApplyAckSender? _ackSender;
private readonly ILoggingAdapter _log = Context.GetLogger(); private readonly ILoggingAdapter _log = Context.GetLogger();
/// <summary>Creates the props for this actor.</summary> /// <summary>Creates the props for this actor.</summary>
/// <param name="centralClient"> /// <param name="centralClient">
/// ClusterClient dialling central's receptionist, or <see langword="null"/> under /// ClusterClient dialling central's receptionist, or <see langword="null"/> under
/// <c>MeshTransport:Mode = Dps</c>, where acks still go out over DistributedPubSub from /// <c>MeshTransport:Mode = Dps</c>, where acks still go out over DistributedPubSub from
/// <c>DriverHostActor</c> and this actor never sees them. /// <c>DriverHostActor</c> and this actor never sees them — and under
/// <c>Mode = Grpc</c>, where no ClusterClient is created at all.
/// </param>
/// <param name="ackSender">
/// The gRPC ack transport under <c>MeshTransport:Mode = Grpc</c>, or <see langword="null"/>
/// under <c>Dps</c> and <c>ClusterClient</c>.
/// </param> /// </param>
/// <returns>The props.</returns> /// <returns>The props.</returns>
public static Props Props(IActorRef? centralClient) => public static Props Props(IActorRef? centralClient, IApplyAckSender? ackSender = null) =>
Akka.Actor.Props.Create(() => new NodeCommunicationActor(centralClient)); Akka.Actor.Props.Create(() => new NodeCommunicationActor(centralClient, ackSender));
/// <summary>Initializes a new instance of the <see cref="NodeCommunicationActor"/> class.</summary> /// <summary>Initializes a new instance of the <see cref="NodeCommunicationActor"/> class.</summary>
/// <param name="centralClient">ClusterClient dialling central, or <see langword="null"/>.</param> /// <param name="centralClient">ClusterClient dialling central, or <see langword="null"/>.</param>
public NodeCommunicationActor(IActorRef? centralClient) /// <param name="ackSender">The gRPC ack transport, or <see langword="null"/>.</param>
public NodeCommunicationActor(IActorRef? centralClient, IApplyAckSender? ackSender = null)
{ {
_centralClient = centralClient; _centralClient = centralClient;
_ackSender = ackSender;
// Inbound from central. Each type is listed explicitly rather than caught by a ReceiveAny: // Inbound from central. Each type is listed explicitly rather than caught by a ReceiveAny:
// an unknown type must dead-letter loudly during development rather than be republished // an unknown type must dead-letter loudly during development rather than be republished
@@ -75,20 +83,54 @@ public sealed class NodeCommunicationActor : ReceiveActor
private void HandleApplyAck(ApplyAck ack) private void HandleApplyAck(ApplyAck ack)
{ {
if (_ackSender is not null)
{
// Fire-and-forget by construction, exactly as the ClusterClient Tell below is. The sender
// never throws and never retries (see IApplyAckSender), so there is nothing to await and
// nothing to pipe back: awaiting would stall this mailbox behind a network call to central,
// and an Ask would invent a request/response the one-way contract does not have.
//
// Checked BEFORE the ClusterClient. The registration arm builds exactly one transport, so
// both are never set — but were they, posting over both would deliver two acks for one
// apply and duplicate central's work for no benefit.
try
{
_ = _ackSender.SendAsync(ack);
}
catch (Exception ex)
{
// "Never throws" is a contract on a public seam, which makes it documentation until
// something enforces it. An implementation that threw synchronously — a guard clause
// ahead of the async state machine is the easy way to do it by accident — would fault
// this Receive, restart the actor, and take the node's whole inbound command path down
// with it. One dropped ack is a deployment that times out; a restarting comm actor is
// every command to this node going missing.
_log.Error(
ex,
"The ApplyAck sender threw synchronously for {DeploymentId}; dropping the ack. "
+ "IApplyAckSender must never throw — this is a defect in the registered sender.",
ack.DeploymentId);
}
return;
}
if (_centralClient is null) if (_centralClient is null)
{ {
_log.Warning( _log.Warning(
"No central ClusterClient — dropping ApplyAck for {DeploymentId}. The ack is not " "No central ack transport — dropping ApplyAck for {DeploymentId}. The ack is not "
+ "buffered and will not be retried, so the deployment will time out at central " + "buffered and will not be retried, so the deployment will time out at central "
+ "naming this node even though it applied successfully. Check " + "naming this node even though it applied successfully. Check "
+ "MeshTransport:CentralContactPoints", + "MeshTransport:CentralContactPoints (ClusterClient mode) or "
+ "MeshTransport:CentralGrpcEndpoints (Grpc mode)",
ack.DeploymentId); ack.DeploymentId);
return; return;
} }
// Send, not SendToAll: the deploy coordinator is a single singleton, and the central comm // Send, not SendToAll: the deploy coordinator is a single singleton, and the central comm
// actor forwards to its proxy. SendToAll would deliver one ack per central node, and the // actor forwards to its proxy. Its NodeId-keyed upsert absorbs a duplicate idempotently,
// coordinator would count the same node twice. // so single-target is defence in depth and a bound on wasted sends — mirroring the
// corrected Grpc-branch rationale above.
_centralClient.Tell(new ClusterClient.Send(MeshPaths.CentralCommunication, ack)); _centralClient.Tell(new ClusterClient.Send(MeshPaths.CentralCommunication, ack));
} }
} }
@@ -374,8 +374,9 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers
/// <param name="dbFactory">Database context factory for configuration database access.</param> /// <param name="dbFactory">Database context factory for configuration database access.</param>
/// <param name="localNode">The local cluster node identifier.</param> /// <param name="localNode">The local cluster node identifier.</param>
/// <param name="ackRouter">Where ApplyAcks are sent. A direct coordinator handle in tests; under /// <param name="ackRouter">Where ApplyAcks are sent. A direct coordinator handle in tests; under
/// <c>MeshTransport:Mode = ClusterClient</c> the node's <c>NodeCommunicationActor</c>, which relays /// <c>MeshTransport:Mode = ClusterClient</c> <b>and</b> <c>= Grpc</c> the node's
/// across the boundary. Null publishes on the <c>deployment-acks</c> DPS topic.</param> /// <c>NodeCommunicationActor</c>, which relays across the boundary over whichever transport it was
/// built with. Null publishes on the <c>deployment-acks</c> DPS topic.</param>
/// <param name="driverFactory">Optional driver factory; defaults to null factory if not provided.</param> /// <param name="driverFactory">Optional driver factory; defaults to null factory if not provided.</param>
/// <param name="localRoles">Optional set of roles assigned to the local node.</param> /// <param name="localRoles">Optional set of roles assigned to the local node.</param>
/// <param name="dependencyMux">Optional actor reference for dependency multiplexing.</param> /// <param name="dependencyMux">Optional actor reference for dependency multiplexing.</param>
@@ -458,8 +459,9 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers
/// <param name="dbFactory">Database context factory for configuration database access.</param> /// <param name="dbFactory">Database context factory for configuration database access.</param>
/// <param name="localNode">The local cluster node identifier.</param> /// <param name="localNode">The local cluster node identifier.</param>
/// <param name="ackRouter">Where ApplyAcks are sent. A direct coordinator handle in tests; under /// <param name="ackRouter">Where ApplyAcks are sent. A direct coordinator handle in tests; under
/// <c>MeshTransport:Mode = ClusterClient</c> the node's <c>NodeCommunicationActor</c>, which relays /// <c>MeshTransport:Mode = ClusterClient</c> <b>and</b> <c>= Grpc</c> the node's
/// across the boundary. Null publishes on the <c>deployment-acks</c> DPS topic.</param> /// <c>NodeCommunicationActor</c>, which relays across the boundary over whichever transport it was
/// built with. Null publishes on the <c>deployment-acks</c> DPS topic.</param>
/// <param name="driverFactory">Optional driver factory; defaults to null factory if not provided.</param> /// <param name="driverFactory">Optional driver factory; defaults to null factory if not provided.</param>
/// <param name="localRoles">Optional set of roles assigned to the local node.</param> /// <param name="localRoles">Optional set of roles assigned to the local node.</param>
/// <param name="dependencyMux">Optional actor reference for dependency multiplexing.</param> /// <param name="dependencyMux">Optional actor reference for dependency multiplexing.</param>
@@ -2490,11 +2492,11 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers
var ack = new ApplyAck(deploymentId, _localNode, outcome, failureReason, correlation); var ack = new ApplyAck(deploymentId, _localNode, outcome, failureReason, correlation);
if (_ackRouter is not null) if (_ackRouter is not null)
{ {
// Either a direct coordinator handle (tests) or, under // Either a direct coordinator handle (tests) or, under MeshTransport:Mode = ClusterClient
// MeshTransport:Mode = ClusterClient, the NodeCommunicationActor that relays this // and = Grpc alike, the NodeCommunicationActor that relays this across the boundary (it
// across the boundary. Renamed from _coordinatorOverride in per-cluster mesh Phase 2: // owns the choice of transport, not this actor). Renamed from _coordinatorOverride in
// under ClusterClient mode this ref is emphatically NOT a coordinator, and the old // per-cluster mesh Phase 2: under those two modes this ref is emphatically NOT a
// name would send the next reader looking for one. // coordinator, and the old name would send the next reader looking for one.
_ackRouter.Tell(ack); _ackRouter.Tell(ack);
} }
else else
@@ -1,6 +1,7 @@
using System.Collections.Immutable; using System.Collections.Immutable;
using Akka.Actor; using Akka.Actor;
using Akka.Cluster.Tools.Client; using Akka.Cluster.Tools.Client;
using Akka.Event;
using Akka.Hosting; using Akka.Hosting;
using Microsoft.EntityFrameworkCore; using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.Configuration; using Microsoft.Extensions.Configuration;
@@ -99,6 +100,26 @@ public static class ServiceCollectionExtensions
// enabled, so piggybacking would leave a redundant pair that historizes values but not alarms // enabled, so piggybacking would leave a redundant pair that historizes values but not alarms
// with no gate at all, and both nodes would write every sample twice with nothing to say so. // with no gate at all, and both nodes would write every sample twice with nothing to say so.
services.TryAddSingleton<IHistorizationRoleView>(sp => sp.GetRequiredService<RedundancyRoleView>()); services.TryAddSingleton<IHistorizationRoleView>(sp => sp.GetRequiredService<RedundancyRoleView>());
// ClusterClient→gRPC migration Task 1B.2: the node→central ack transport under
// MeshTransport:Mode = Grpc. Registered unconditionally but resolved only by the Grpc arm of
// WithOtOpcUaRuntimeActors — the mode is not knowable here (this runs before configuration is
// read into options), and a singleton nobody resolves is never constructed, so a Dps node
// opens no channel.
//
// Registered in the CONTAINER rather than new-ed up beside the actor so its per-endpoint
// GrpcChannel cache is disposed at host shutdown; an actor-scoped instance would leak the
// channels on every restart.
//
// Built by a factory, not by type: MS.DI does not honour optional constructor parameters, so
// the clientFactory test seam would make type-based activation fail outright. The logger is
// resolved leniently for the same reason DriverHostActor's is — a harness that skipped logging
// should not fail to build the graph.
services.TryAddSingleton<IApplyAckSender>(sp => new GrpcApplyAckSender(
sp.GetRequiredService<IOptions<MeshTransportOptions>>(),
sp.GetRequiredService<IOptions<AkkaClusterOptions>>(),
sp.GetService<ILoggerFactory>()?.CreateLogger<GrpcApplyAckSender>()
?? NullLogger<GrpcApplyAckSender>.Instance));
return services; return services;
} }
@@ -389,6 +410,8 @@ public static class ServiceCollectionExtensions
var meshOptions = resolver.GetService<IOptions<MeshTransportOptions>>().Value; var meshOptions = resolver.GetService<IOptions<MeshTransportOptions>>().Value;
var useClusterClient = string.Equals( var useClusterClient = string.Equals(
meshOptions.Mode, MeshTransportOptions.ModeClusterClient, StringComparison.OrdinalIgnoreCase); meshOptions.Mode, MeshTransportOptions.ModeClusterClient, StringComparison.OrdinalIgnoreCase);
var useGrpc = string.Equals(
meshOptions.Mode, MeshTransportOptions.ModeGrpc, StringComparison.OrdinalIgnoreCase);
IActorRef? centralClient = null; IActorRef? centralClient = null;
if (useClusterClient) if (useClusterClient)
@@ -406,12 +429,33 @@ public static class ServiceCollectionExtensions
CentralClusterClientName); CentralClusterClientName);
} }
// ClusterClient→gRPC migration Task 1B.2. Deliberately EXCLUSIVE with the branch above —
// under Grpc no ClusterClient is created at all, which is half the point of the migration
// (it is the cross-WAN Akka association this program removes). The two are never both
// handed to the actor, and it prefers the sender if they ever were.
IApplyAckSender? ackSender = null;
if (useGrpc)
{
ackSender = resolver.GetService<IApplyAckSender>();
if (ackSender is null)
{
// Reachable only in a graph that skipped AddOtOpcUaRuntime (a test harness), never
// in the Host. Said here rather than left to the actor's per-ack drop warning,
// because the cause is a missing registration and the symptom would otherwise
// appear once per deployment with no hint of why.
system.Log.Warning(
"MeshTransport:Mode is {Mode} but no {Sender} is registered — every ApplyAck "
+ "this node produces will be dropped. AddOtOpcUaRuntime registers it.",
meshOptions.Mode, nameof(IApplyAckSender));
}
}
var nodeComm = system.ActorOf( var nodeComm = system.ActorOf(
NodeCommunicationActor.Props(centralClient), NodeCommunicationActor.Props(centralClient, ackSender),
MeshPaths.NodeCommunicationName); MeshPaths.NodeCommunicationName);
// Registered with the receptionist in BOTH modes: under Dps nothing dials it, and an // Registered with the receptionist in ALL modes: under Dps and Grpc nothing dials it, and
// idle registration costs nothing — whereas registering only under ClusterClient would // an idle registration costs nothing — whereas registering only under ClusterClient would
// make flipping the flag require a node restart before central could reach it. // make flipping the flag require a node restart before central could reach it.
ClusterClientReceptionist.Get(system).RegisterService(nodeComm); ClusterClientReceptionist.Get(system).RegisterService(nodeComm);
registry.Register<NodeCommunicationActorKey>(nodeComm); registry.Register<NodeCommunicationActorKey>(nodeComm);
@@ -523,9 +567,12 @@ public static class ServiceCollectionExtensions
: null; : null;
var driverHost = system.ActorOf( var driverHost = system.ActorOf(
// ackRouter: under ClusterClient mode the node comm actor relays ApplyAcks across // ackRouter: under ClusterClient AND Grpc the node comm actor relays ApplyAcks across
// the boundary; under Dps it stays null and the host publishes on the // the boundary (which transport it then uses is its own business — the ClusterClient
// deployment-acks topic exactly as before. // ref or the gRPC sender it was built with); under Dps it stays null and the host
// publishes on the deployment-acks topic exactly as before. Grpc MUST be included
// here: routed to DPS instead, a Grpc node's ack would never reach the sender at all
// and every deployment to it would time out while it applied perfectly.
// dbFactory is nullable (Phase 4): a driver-only node passes null here — ConfigDb is // dbFactory is nullable (Phase 4): a driver-only node passes null here — ConfigDb is
// gated on hasAdmin and driver-only nodes are forced to FetchAndCache. DriverHostActor // gated on hasAdmin and driver-only nodes are forced to FetchAndCache. DriverHostActor
// guards every ConfigDb-touching path on it (UpsertNodeDeploymentState no-ops; central // guards every ConfigDb-touching path on it (UpsertNodeDeploymentState no-ops; central
@@ -533,7 +580,7 @@ public static class ServiceCollectionExtensions
// alarmStateStore (there is no ConfigDb-backed EF fallback; the host skips the alarm // alarmStateStore (there is no ConfigDb-backed EF fallback; the host skips the alarm
// host when no store was wired). // host when no store was wired).
DriverHostActor.Props(dbFactory, roleInfo.LocalNode, DriverHostActor.Props(dbFactory, roleInfo.LocalNode,
ackRouter: useClusterClient ? nodeComm : null, ackRouter: useClusterClient || useGrpc ? nodeComm : null,
driverFactory: driverFactory, localRoles: roleInfo.LocalRoles, driverFactory: driverFactory, localRoles: roleInfo.LocalRoles,
dependencyMux: mux, dependencyMux: mux,
opcUaPublishActor: publishActor, opcUaPublishActor: publishActor,
@@ -20,9 +20,18 @@ namespace ZB.MOM.WW.OtOpcUa.Cluster.Tests;
[Collection(RoleEnvironmentCollection.Name)] [Collection(RoleEnvironmentCollection.Name)]
public class MeshTransportOptionsValidatorTests public class MeshTransportOptionsValidatorTests
{ {
private static ValidateOptionsResult Validate(MeshTransportOptions o, params string[] roles) private static ValidateOptionsResult Validate(MeshTransportOptions o, params string[] roles) =>
Validate(o, new Dictionary<string, string?>(), roles);
/// <summary>
/// As <see cref="Validate(MeshTransportOptions, string[])"/>, plus arbitrary configuration
/// keys — the admin-side rule cross-reads <c>ConfigServe:GrpcListenPort</c>, which lives
/// outside this options class entirely.
/// </summary>
private static ValidateOptionsResult Validate(
MeshTransportOptions o, Dictionary<string, string?> extra, params string[] roles)
{ {
var pairs = new Dictionary<string, string?>(); var pairs = new Dictionary<string, string?>(extra);
for (var i = 0; i < roles.Length; i++) for (var i = 0; i < roles.Length; i++)
{ {
pairs[$"Cluster:Roles:{i}"] = roles[i]; pairs[$"Cluster:Roles:{i}"] = roles[i];
@@ -34,6 +43,9 @@ public class MeshTransportOptionsValidatorTests
.Validate(MeshTransportOptions.SectionName, o); .Validate(MeshTransportOptions.SectionName, o);
} }
private static Dictionary<string, string?> ConfigServePort(string? value) =>
new() { ["ConfigServe:GrpcListenPort"] = value };
private static MeshTransportOptions WellFormedGrpc() => new() private static MeshTransportOptions WellFormedGrpc() => new()
{ {
Mode = MeshTransportOptions.ModeGrpc, Mode = MeshTransportOptions.ModeGrpc,
@@ -298,11 +310,105 @@ public class MeshTransportOptionsValidatorTests
// The node-side surface is what a DRIVER uses to dial central and post its ack. An admin-only // The node-side surface is what a DRIVER uses to dial central and post its ack. An admin-only
// central node dials nodes instead and resolves per-cluster keys at runtime, so requiring // central node dials nodes instead and resolves per-cluster keys at runtime, so requiring
// these here would make an admin node carry a PSK and an endpoint list it never reads. // these here would make an admin node carry a PSK and an endpoint list it never reads.
// (The ConfigServe port IS required of an admin node under Grpc — supplied here so this test
// stays about the node-side keys and nothing else.)
Validate( Validate(
new MeshTransportOptions { Mode = MeshTransportOptions.ModeGrpc }, new MeshTransportOptions { Mode = MeshTransportOptions.ModeGrpc },
ConfigServePort("4055"),
"admin", "cluster-MAIN").Succeeded.ShouldBeTrue(); "admin", "cluster-MAIN").Succeeded.ShouldBeTrue();
} }
[Fact]
public void Grpc_admin_node_without_a_config_serve_listener_fails()
{
// Central hosts CentralCommandService on the ConfigServe h2c listener, so an unset port means
// the service is never mapped and every node's ack has nowhere to land. Nothing errors at
// runtime — the deployment simply times out naming nodes that applied perfectly, which is the
// exact absence this validator exists to convert into a refusal to start.
var result = Validate(
new MeshTransportOptions { Mode = MeshTransportOptions.ModeGrpc },
"admin", "cluster-MAIN");
result.Failed.ShouldBeTrue();
result.FailureMessage.ShouldContain("ConfigServe:GrpcListenPort");
result.FailureMessage.ShouldContain("ConfigServe__GrpcListenPort");
}
[Fact]
public void Grpc_admin_node_with_a_zero_config_serve_port_fails()
{
// 0 is the documented "disabled" value, not merely a missing key — it must fail exactly as an
// absent one does, because it binds nothing just as surely.
Validate(
new MeshTransportOptions { Mode = MeshTransportOptions.ModeGrpc },
ConfigServePort("0"),
"admin", "cluster-MAIN").Failed.ShouldBeTrue();
}
[Fact]
public void Grpc_admin_node_with_an_unparseable_config_serve_port_fails()
{
// A typo'd port binds nothing, so it belongs on the same side of the line as an absent one.
Validate(
new MeshTransportOptions { Mode = MeshTransportOptions.ModeGrpc },
ConfigServePort("40o55"),
"admin", "cluster-MAIN").Failed.ShouldBeTrue();
}
[Fact]
public void Grpc_admin_node_with_a_config_serve_listener_passes()
{
Validate(
new MeshTransportOptions { Mode = MeshTransportOptions.ModeGrpc },
ConfigServePort("4055"),
"admin", "cluster-MAIN").Succeeded.ShouldBeTrue();
}
[Fact]
public void Grpc_driver_only_node_needs_no_config_serve_listener()
{
// A driver node dials central; it serves no central command endpoint at all, and requiring an
// admin-side listener port of it would demand configuration it never reads.
Validate(WellFormedGrpc(), "driver", "cluster-SITE-A").Succeeded.ShouldBeTrue();
}
[Fact]
public void Dps_admin_node_needs_no_config_serve_listener()
{
// The no-new-obligations invariant, admin side: a central node still on the default transport
// publishes on DistributedPubSub and hosts no command endpoint.
Validate(
new MeshTransportOptions { Mode = MeshTransportOptions.ModeDps },
"admin", "cluster-MAIN").Succeeded.ShouldBeTrue();
}
[Fact]
public void ClusterClient_admin_node_needs_no_config_serve_listener()
{
// Contact points are required by the mode itself, independently of role — supplied so this
// test fails only if the ConfigServe rule wrongly fires outside Grpc mode.
Validate(
new MeshTransportOptions
{
Mode = MeshTransportOptions.ModeClusterClient,
CentralContactPoints = ["akka.tcp://otopcua@central-1:4053"],
},
"admin", "cluster-MAIN").Succeeded.ShouldBeTrue();
}
[Fact]
public void The_admin_role_is_also_read_from_OTOPCUA_ROLES()
{
// Same roles-source divergence as the driver case (Gitea #532), on the side that hosts the
// endpoint: a central node configured only through the env var is still the node whose
// listener every ack depends on.
using (new EnvVarScope("OTOPCUA_ROLES", "admin,cluster-MAIN"))
{
Validate(new MeshTransportOptions { Mode = MeshTransportOptions.ModeGrpc })
.Failed.ShouldBeTrue();
}
}
[Fact] [Fact]
public void Grpc_role_less_node_needs_none_of_the_node_side_keys() public void Grpc_role_less_node_needs_none_of_the_node_side_keys()
{ {
@@ -348,6 +454,29 @@ public class MeshTransportOptionsValidatorTests
"driver", "cluster-SITE-A").Succeeded.ShouldBeTrue(); "driver", "cluster-SITE-A").Succeeded.ShouldBeTrue();
} }
[Fact]
public void Grpc_fused_admin_and_driver_node_must_satisfy_BOTH_surfaces()
{
// The central pair is fused (admin,driver,cluster-MAIN), so it is the one node shape that owes
// both halves at once — and the two blocks are independent `if`s, not an `if`/`else if`. Were
// they chained, a fused node would be graded on the driver surface only and its missing
// ConfigServe listener would pass validation, leaving central with no place for any node's ack
// to land. Asserted from BOTH sides so a future chaining in either order is caught.
var missingNodeSide = new MeshTransportOptions { Mode = MeshTransportOptions.ModeGrpc };
var withPort = Validate(missingNodeSide, ConfigServePort("4055"), "driver", "admin", "cluster-MAIN");
withPort.Failed.ShouldBeTrue();
withPort.FailureMessage.ShouldContain("MeshTransport:GrpcPsk");
withPort.FailureMessage.ShouldContain("MeshTransport:CentralGrpcEndpoints");
var missingAdminSide = Validate(WellFormedGrpc(), "driver", "admin", "cluster-MAIN");
missingAdminSide.Failed.ShouldBeTrue();
missingAdminSide.FailureMessage.ShouldContain("ConfigServe:GrpcListenPort");
// And the fully-configured fused node passes — the rule is "both", not "neither".
Validate(WellFormedGrpc(), ConfigServePort("4055"), "driver", "admin", "cluster-MAIN")
.Succeeded.ShouldBeTrue();
}
[Fact] [Fact]
public void ClusterPsks_defaults_to_an_empty_case_insensitive_map() public void ClusterPsks_defaults_to_an_empty_case_insensitive_map()
{ {
@@ -0,0 +1,266 @@
using Akka.Hosting;
using Akka.TestKit.Xunit2;
using Grpc.Core;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Logging.Abstractions;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.AdminUI.Grpc;
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Deploy;
using ZB.MOM.WW.OtOpcUa.Commons.Protos.MeshTransport.V1;
using ZB.MOM.WW.OtOpcUa.Commons.Types;
using ZB.MOM.WW.OtOpcUa.ControlPlane;
namespace ZB.MOM.WW.OtOpcUa.AdminUI.Tests.Grpc;
/// <summary>
/// The central-side gRPC ingress for a node's deployment ack (ClusterClient→gRPC migration
/// Phase 1B). It decodes the wire message and hands it to the registered
/// <c>CentralCommunicationActor</c> — the same actor the ClusterClient receptionist delivered to,
/// whose forward-to-the-coordinator path is untouched.
/// </summary>
/// <remarks>
/// Both failure modes matter more than the happy path. An ack that cannot be decoded must be
/// refused rather than guessed at (a wrong outcome seals a deployment nobody applied), and an ack
/// arriving before the actor exists must say <see cref="StatusCode.Unavailable"/> so the caller
/// retries instead of recording a permanent failure against a node that is fine.
/// </remarks>
public sealed class CentralCommandGrpcServiceTests : TestKit
{
private static ApplyAckMessage WellFormed() => new()
{
DeploymentId = Guid.NewGuid().ToString("N"),
NodeId = "site-a-1",
Outcome = ApplyOutcome.Applied,
CorrelationId = Guid.NewGuid().ToString("N"),
};
private static CentralCommandGrpcService Service(ActorRegistry registry) =>
new(registry, NullLogger<CentralCommandGrpcService>.Instance);
[Fact]
public async Task A_well_formed_ack_is_decoded_and_told_to_the_registered_actor()
{
var probe = CreateTestProbe("central-comm");
var registry = new ActorRegistry();
registry.Register<CentralCommunicationActorKey>(probe.Ref);
var request = WellFormed();
request.FailureReason = "disk full";
request.Outcome = ApplyOutcome.Failed;
var response = await Service(registry).ReportApply(request, new FakeServerCallContext());
response.ShouldNotBeNull();
var received = probe.ExpectMsg<ApplyAck>(TimeSpan.FromSeconds(3));
received.DeploymentId.ShouldBe(DeploymentId.Parse(request.DeploymentId));
received.NodeId.Value.ShouldBe("site-a-1");
received.Outcome.ShouldBe(ApplyAckOutcome.Failed);
received.FailureReason.ShouldBe("disk full");
received.CorrelationId.ToString().ShouldBe(request.CorrelationId);
}
[Fact]
public async Task An_absent_failure_reason_arrives_as_null_and_an_empty_one_as_empty()
{
// Presence, not emptiness — the mapper's contract, asserted at the seam that actually feeds
// the coordinator. Collapsing the two would report an operator-visible blank reason as "no
// reason recorded".
var probe = CreateTestProbe("central-comm");
var registry = new ActorRegistry();
registry.Register<CentralCommunicationActorKey>(probe.Ref);
var service = Service(registry);
await service.ReportApply(WellFormed(), new FakeServerCallContext());
probe.ExpectMsg<ApplyAck>(TimeSpan.FromSeconds(3)).FailureReason.ShouldBeNull();
var blank = WellFormed();
blank.FailureReason = "";
await service.ReportApply(blank, new FakeServerCallContext());
probe.ExpectMsg<ApplyAck>(TimeSpan.FromSeconds(3)).FailureReason.ShouldBe("");
}
[Fact]
public async Task An_unregistered_actor_yields_Unavailable()
{
// The boot window: the listener answers before Akka.Hosting has spawned the actor. Unavailable
// is the retryable status, so the node's next attempt lands instead of the deployment
// recording a permanent failure against a node that applied fine.
var ex = await Should.ThrowAsync<RpcException>(() =>
Service(new ActorRegistry()).ReportApply(WellFormed(), new FakeServerCallContext()));
ex.StatusCode.ShouldBe(StatusCode.Unavailable);
}
[Fact]
public async Task An_unspecified_outcome_yields_InvalidArgument()
{
// proto3 decodes an absent enum as 0. Refusing is the whole point: the deploy path seals on
// Applied, so guessing an outcome the node never sent would seal a deployment nobody applied.
var probe = CreateTestProbe("central-comm");
var registry = new ActorRegistry();
registry.Register<CentralCommunicationActorKey>(probe.Ref);
var request = WellFormed();
request.Outcome = ApplyOutcome.Unspecified;
var ex = await Should.ThrowAsync<RpcException>(() =>
Service(registry).ReportApply(request, new FakeServerCallContext()));
ex.StatusCode.ShouldBe(StatusCode.InvalidArgument);
probe.ExpectNoMsg(TimeSpan.FromMilliseconds(300));
}
[Fact]
public async Task A_malformed_id_yields_InvalidArgument()
{
var request = WellFormed();
request.DeploymentId = "not-a-guid";
var probe = CreateTestProbe("central-comm");
var registry = new ActorRegistry();
registry.Register<CentralCommunicationActorKey>(probe.Ref);
var ex = await Should.ThrowAsync<RpcException>(() =>
Service(registry).ReportApply(request, new FakeServerCallContext()));
ex.StatusCode.ShouldBe(StatusCode.InvalidArgument);
probe.ExpectNoMsg(TimeSpan.FromMilliseconds(300));
}
[Fact]
public async Task A_blank_node_id_yields_InvalidArgument()
{
// NodeId.Parse refuses a blank name; an ack whose node cannot be named is uncountable at the
// coordinator, so it must be refused here rather than forwarded as an anonymous ack.
var request = WellFormed();
request.NodeId = "";
var probe = CreateTestProbe("central-comm");
var registry = new ActorRegistry();
registry.Register<CentralCommunicationActorKey>(probe.Ref);
var ex = await Should.ThrowAsync<RpcException>(() =>
Service(registry).ReportApply(request, new FakeServerCallContext()));
ex.StatusCode.ShouldBe(StatusCode.InvalidArgument);
probe.ExpectNoMsg(TimeSpan.FromMilliseconds(300));
}
[Fact]
public async Task An_undecodable_message_is_graded_before_the_registry()
{
// Ordering pin. With NO actor registered AND an undecodable message, the answer is
// InvalidArgument, not Unavailable: Unavailable invites a retry, and retrying a message that
// can never decode is a loop. The caller's fault is reported as the caller's fault.
var request = WellFormed();
request.Outcome = ApplyOutcome.Unspecified;
var ex = await Should.ThrowAsync<RpcException>(() =>
Service(new ActorRegistry()).ReportApply(request, new FakeServerCallContext()));
ex.StatusCode.ShouldBe(StatusCode.InvalidArgument);
}
[Fact]
public async Task A_decode_failure_logs_the_raw_correlation_id()
{
// The correlation id is logged as the raw wire string precisely because the ids may be what
// failed to parse — an operator correlating a rejection cannot be given a value that only
// exists once decoding succeeded.
var logger = new CapturingLogger<CentralCommandGrpcService>();
var request = WellFormed();
request.CorrelationId = "correlation-that-never-parses";
request.DeploymentId = "not-a-guid";
await Should.ThrowAsync<RpcException>(() =>
new CentralCommandGrpcService(new ActorRegistry(), logger)
.ReportApply(request, new FakeServerCallContext()));
logger.Entries.Any(e => e.Message.Contains("correlation-that-never-parses", StringComparison.Ordinal))
.ShouldBeTrue("the rejection log must carry the raw correlation id");
}
[Fact]
public async Task A_not_ready_rejection_logs_the_peer_and_the_correlation_id()
{
// Diagnostic symmetry with the decode-failure warning above. The two rejections are the only
// two ways an ack fails to reach the coordinator, and an operator reading the central log has
// to correlate BOTH against the node's own log — which needs the correlation id — and against
// which node was calling, which needs the peer. Decoding has already succeeded on this path,
// so the decoded correlation id costs nothing to include.
var logger = new CapturingLogger<CentralCommandGrpcService>();
var request = WellFormed();
await Should.ThrowAsync<RpcException>(() =>
new CentralCommandGrpcService(new ActorRegistry(), logger)
.ReportApply(request, new FakeServerCallContext()));
var warning = logger.Entries.Single(e => e.Level == LogLevel.Warning);
warning.Message.ShouldContain("ipv4:127.0.0.1:0");
warning.Message.ShouldContain(request.CorrelationId);
}
/// <summary>Test logger that captures all log entries.</summary>
/// <typeparam name="T">The category type.</typeparam>
private sealed class CapturingLogger<T> : ILogger<T>
{
/// <summary>Gets the captured log entries.</summary>
public List<(LogLevel Level, string Message, Exception? Exception)> Entries { get; } = [];
/// <summary>Begins a scope (stub implementation).</summary>
/// <typeparam name="TState">The type of the scope state.</typeparam>
/// <param name="state">The scope state.</param>
/// <returns>A no-op scope.</returns>
public IDisposable BeginScope<TState>(TState state) where TState : notnull => NullScope.Instance;
/// <summary>Checks if logging is enabled (always true).</summary>
/// <param name="logLevel">The log level to check.</param>
/// <returns><c>true</c>.</returns>
public bool IsEnabled(LogLevel logLevel) => true;
/// <summary>Logs an entry and captures it.</summary>
/// <typeparam name="TState">The type of the log state.</typeparam>
/// <param name="logLevel">The log level.</param>
/// <param name="eventId">The event ID.</param>
/// <param name="state">The log state.</param>
/// <param name="exception">The exception, if any.</param>
/// <param name="formatter">The log message formatter.</param>
public void Log<TState>(
LogLevel logLevel, EventId eventId, TState state, Exception? exception,
Func<TState, Exception?, string> formatter)
{
Entries.Add((logLevel, formatter(state, exception), exception));
}
private sealed class NullScope : IDisposable
{
/// <summary>Gets the singleton instance.</summary>
public static NullScope Instance { get; } = new();
/// <summary>No-op.</summary>
public void Dispose() { }
}
}
/// <summary>Minimal <see cref="ServerCallContext"/> — the service reads only the method path.</summary>
private sealed class FakeServerCallContext : ServerCallContext
{
protected override string MethodCore => "/mesh_transport.v1.CentralCommandService/ReportApply";
protected override string HostCore => "localhost";
protected override string PeerCore => "ipv4:127.0.0.1:0";
protected override DateTime DeadlineCore => DateTime.MaxValue;
protected override Metadata RequestHeadersCore => [];
protected override CancellationToken CancellationTokenCore => CancellationToken.None;
protected override Metadata ResponseTrailersCore { get; } = [];
protected override Status StatusCore { get; set; }
protected override WriteOptions? WriteOptionsCore { get; set; }
protected override AuthContext AuthContextCore => new(null, new Dictionary<string, List<AuthProperty>>());
protected override ContextPropagationToken CreatePropagationTokenCore(ContextPropagationOptions? options) =>
throw new NotSupportedException();
protected override Task WriteResponseHeadersAsyncCore(Metadata responseHeaders) => Task.CompletedTask;
}
}
@@ -0,0 +1,424 @@
using Grpc.Core;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Logging.Abstractions;
using Microsoft.Extensions.Options;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Cluster;
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Deploy;
using ZB.MOM.WW.OtOpcUa.Commons.Protos.MeshTransport.V1;
using ZB.MOM.WW.OtOpcUa.Commons.Types;
using ZB.MOM.WW.OtOpcUa.Runtime.Communication;
namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Communication;
/// <summary>
/// The node-side gRPC ack sender (ClusterClient→gRPC migration Task 1B.2). The client is a fake,
/// so what is under test is the sender's own logic: the ordered single-pass over
/// <c>CentralGrpcEndpoints</c>, the per-call credentials, and the drop that used to be silent.
/// </summary>
/// <remarks>
/// <para>
/// <b>The load-bearing fact is that ONE endpoint is dialled on success.</b> This replaces an
/// Akka <c>ClusterClient.Send</c>, which is single-target at-most-once — not a
/// <c>SendToAll</c>. Both central nodes route an ack to the same
/// <c>ConfigPublishCoordinator</c> singleton proxy, whose NodeId-keyed upsert absorbs a
/// duplicate idempotently — so the single-endpoint rule is defence in depth and a bound
/// on wasted dials, not what protects the seal.
/// </para>
/// <para>
/// The failure side matters just as much: the receptionist dropped an undeliverable ack in
/// silence, and the whole point of the migration is that the same drop is now a Warning naming
/// every endpoint and the status it answered with.
/// </para>
/// </remarks>
public sealed class GrpcApplyAckSenderTests
{
private const string Psk = "psk-that-must-never-be-logged";
private static ApplyAck Ack() => new(
new DeploymentId(Guid.NewGuid()),
new NodeId("site-a-1:4053"),
ApplyAckOutcome.Applied,
null,
new CorrelationId(Guid.NewGuid()));
private static GrpcApplyAckSender Sender(
string[] endpoints,
Func<string, CentralCommandService.CentralCommandServiceClient> factory,
ILogger<GrpcApplyAckSender>? log = null,
string[]? roles = null)
=> new(
Options.Create(new MeshTransportOptions
{
Mode = MeshTransportOptions.ModeGrpc,
CentralGrpcEndpoints = endpoints,
GrpcPsk = Psk,
AckTimeoutSeconds = 7,
}),
Options.Create(new AkkaClusterOptions { Roles = roles ?? ["driver", "cluster-SITE-A"] }),
log ?? NullLogger<GrpcApplyAckSender>.Instance,
factory);
[Fact]
public async Task A_successful_first_endpoint_is_the_ONLY_one_dialled()
{
// The single-count semantic. Two endpoints are configured for availability, not for fan-out:
// an ack posted to both would reach the one coordinator singleton twice.
var consulted = new List<string>();
var calls = new List<(ApplyAckMessage Request, CallOptions Options)>();
var factory = (string endpoint) =>
{
consulted.Add(endpoint);
return (CentralCommandService.CentralCommandServiceClient)new RecordingClient(calls);
};
await Sender(["http://central-1:4055", "http://central-2:4055"], factory).SendAsync(Ack());
consulted.ShouldBe(["http://central-1:4055"]);
calls.Count.ShouldBe(1);
}
[Fact]
public async Task The_wire_message_carries_the_whole_ack()
{
var calls = new List<(ApplyAckMessage Request, CallOptions Options)>();
var ack = new ApplyAck(
new DeploymentId(Guid.NewGuid()),
new NodeId("site-a-2:4053"),
ApplyAckOutcome.Failed,
"disk full",
new CorrelationId(Guid.NewGuid()));
await Sender(["http://central-1:4055"], _ => new RecordingClient(calls)).SendAsync(ack);
var sent = calls.Single().Request;
sent.DeploymentId.ShouldBe(ack.DeploymentId.ToString());
sent.NodeId.ShouldBe("site-a-2:4053");
sent.Outcome.ShouldBe(ApplyOutcome.Failed);
sent.FailureReason.ShouldBe("disk full");
sent.CorrelationId.ShouldBe(ack.CorrelationId.ToString());
}
[Fact]
public async Task Every_call_carries_the_bearer_the_cluster_header_and_a_deadline()
{
// Central's CentralCommandAuthInterceptor grades the bearer against the key of the cluster
// named in x-otopcua-cluster, so an ack missing either header is PermissionDenied — and the
// deadline is what stops a dead first endpoint from stalling the ack forever.
var calls = new List<(ApplyAckMessage Request, CallOptions Options)>();
var before = DateTime.UtcNow;
await Sender(["http://central-1:4055"], _ => new RecordingClient(calls)).SendAsync(Ack());
var options = calls.Single().Options;
options.Headers.ShouldNotBeNull();
options.Headers.GetValue("authorization").ShouldBe($"Bearer {Psk}");
options.Headers.GetValue("x-otopcua-cluster").ShouldBe("SITE-A");
options.Deadline.ShouldNotBeNull();
// Bounded from BOTH sides against the configured 7s. A lower bound of `before` alone would
// accept any positive timeout at all — the value could be ignored entirely and the fact would
// still pass.
options.Deadline!.Value.ShouldBeGreaterThan(before.AddSeconds(6));
options.Deadline!.Value.ShouldBeLessThanOrEqualTo(before.AddSeconds(8));
}
[Fact]
public async Task Each_endpoint_gets_a_FRESH_deadline()
{
// The failure this pins is silent and is the characteristic WAN one: a central that HANGS
// rather than refusing. With one absolute deadline hoisted over the whole walk, endpoint 1
// burns all of AckTimeoutSeconds, and endpoint 2 is then dialled with a deadline already in
// the past — it fails instantly as DeadlineExceeded without a packet leaving the node, so the
// surviving central is never really tried and failover is defeated while still LOOKING like it
// ran. Both stamps come from real UtcNow, so a delay on the failing path plus a strict `>` is
// the whole test.
var calls = new List<(ApplyAckMessage Request, CallOptions Options)>();
var factory = (string endpoint) => endpoint.Contains("central-1", StringComparison.Ordinal)
? (CentralCommandService.CentralCommandServiceClient)new SlowThrowingClient(calls, TimeSpan.FromMilliseconds(50))
: new RecordingClient(calls);
await Sender(["http://central-1:4055", "http://central-2:4055"], factory).SendAsync(Ack());
calls.Count.ShouldBe(2);
calls[1].Options.Deadline!.Value.ShouldBeGreaterThan(calls[0].Options.Deadline!.Value);
}
[Fact]
public async Task An_unavailable_first_endpoint_fails_over_to_the_second()
{
var consulted = new List<string>();
var calls = new List<(ApplyAckMessage Request, CallOptions Options)>();
var factory = (string endpoint) =>
{
consulted.Add(endpoint);
return endpoint.Contains("central-1", StringComparison.Ordinal)
? (CentralCommandService.CentralCommandServiceClient)
new ThrowingClient(new RpcException(new Status(StatusCode.Unavailable, "down")))
: new RecordingClient(calls);
};
await Sender(["http://central-1:4055", "http://central-2:4055"], factory).SendAsync(Ack());
consulted.ShouldBe(["http://central-1:4055", "http://central-2:4055"]);
calls.Count.ShouldBe(1);
}
[Fact]
public async Task When_every_endpoint_fails_the_ack_is_dropped_with_one_naming_Warning()
{
// The pre-migration behaviour was a silent drop. It is still a drop — there is no queue and no
// retry — but the operator now gets the node, the deployment, the correlation id and what each
// endpoint actually answered, which is everything needed to tell "central is down" from
// "central rejected this node's key".
var log = new CapturingLogger<GrpcApplyAckSender>();
var attempts = 0;
var factory = (string endpoint) =>
{
attempts++;
return (CentralCommandService.CentralCommandServiceClient)new ThrowingClient(
new RpcException(new Status(
endpoint.Contains("central-1", StringComparison.Ordinal)
? StatusCode.Unavailable
: StatusCode.PermissionDenied,
"no")));
};
var ack = Ack();
await Sender(["http://central-1:4055", "http://central-2:4055"], factory, log).SendAsync(ack);
attempts.ShouldBe(2);
var warning = log.Entries.Single(e => e.Level == LogLevel.Warning);
warning.Message.ShouldContain(ack.NodeId.Value!);
warning.Message.ShouldContain(ack.DeploymentId.ToString());
warning.Message.ShouldContain(ack.CorrelationId.ToString());
warning.Message.ShouldContain("Unavailable");
warning.Message.ShouldContain("PermissionDenied");
}
[Fact]
public async Task A_failed_pass_never_posts_the_ack_twice()
{
// The counterpart of the first fact: failover walks forward only. A retry of an
// already-delivered endpoint could double-count this node at the coordinator, which is the one
// outcome worse than the ack being lost.
var calls = new List<(ApplyAckMessage Request, CallOptions Options)>();
var consulted = new List<string>();
var factory = (string endpoint) =>
{
consulted.Add(endpoint);
return endpoint.Contains("c1", StringComparison.Ordinal)
? (CentralCommandService.CentralCommandServiceClient)
new ThrowingClient(new RpcException(new Status(StatusCode.Unavailable, "down")))
: new RecordingClient(calls);
};
await Sender(["http://c1:4055", "http://c2:4055", "http://c3:4055"], factory).SendAsync(Ack());
// c3 is never dialled: the pass stops at the first success, and it does not go back to c1.
consulted.ShouldBe(["http://c1:4055", "http://c2:4055"]);
calls.Count.ShouldBe(1);
}
[Fact]
public async Task The_pre_shared_key_is_never_written_to_a_log()
{
// The bearer travels in the call headers on every attempt; the failure log names endpoints and
// status codes only. A per-endpoint dump of the Metadata would put the cluster's PSK in the
// node log of every deployment that could not reach central.
var log = new CapturingLogger<GrpcApplyAckSender>();
var factory = (string _) => (CentralCommandService.CentralCommandServiceClient)
new ThrowingClient(new RpcException(new Status(StatusCode.Unavailable, "down")));
await Sender(["http://central-1:4055"], factory, log).SendAsync(Ack());
log.Entries.ShouldNotBeEmpty();
log.Entries.Any(e => e.Message.Contains(Psk, StringComparison.Ordinal)).ShouldBeFalse();
}
[Fact]
public async Task An_empty_endpoint_list_is_a_logged_drop_not_a_throw()
{
// The validator refuses to start a Grpc driver node with no endpoints, so this is reachable
// only through a code path that built the sender for a node the validator never graded (a test
// harness, an admin-only graph). It must still be a visible drop rather than an exception on a
// task the actor discards, which would vanish entirely.
var log = new CapturingLogger<GrpcApplyAckSender>();
var dialled = false;
var factory = (string _) =>
{
dialled = true;
return (CentralCommandService.CentralCommandServiceClient)new RecordingClient([]);
};
await Sender([], factory, log).SendAsync(Ack());
dialled.ShouldBeFalse();
log.Entries.Single(e => e.Level == LogLevel.Warning).Message
.ShouldContain("CentralGrpcEndpoints");
}
[Fact]
public async Task A_node_with_no_cluster_role_still_posts_and_is_warned_about()
{
// No cluster-scoped role means no x-otopcua-cluster header, and central has no key to grade the
// bearer against — every ack will come back PermissionDenied. Posting anyway (rather than
// refusing locally) keeps the rejection where it can be read on both sides, and the local
// Warning explains a PermissionDenied that would otherwise look like a key mismatch.
var log = new CapturingLogger<GrpcApplyAckSender>();
var calls = new List<(ApplyAckMessage Request, CallOptions Options)>();
await Sender(["http://central-1:4055"], _ => new RecordingClient(calls), log, roles: ["driver"])
.SendAsync(Ack());
calls.Single().Options.Headers!.Any(h =>
h.Key.Equals("x-otopcua-cluster", StringComparison.Ordinal)).ShouldBeFalse();
log.Entries.Any(e => e.Level == LogLevel.Warning
&& e.Message.Contains("cluster-", StringComparison.Ordinal)).ShouldBeTrue();
}
[Fact]
public async Task An_unexpected_client_fault_is_a_logged_drop_not_an_unobserved_task_fault()
{
// The caller is an actor that discards the returned Task, so anything escaping here would be
// swallowed by the runtime and rediscovered as an unobserved exception at GC — the exact
// silent drop this transport exists to remove.
var log = new CapturingLogger<GrpcApplyAckSender>();
var factory = (string _) => (CentralCommandService.CentralCommandServiceClient)
new FaultingClient(new InvalidOperationException("channel disposed"));
await Should.NotThrowAsync(() => Sender(["http://central-1:4055"], factory, log).SendAsync(Ack()));
log.Entries.Any(e => e.Level == LogLevel.Warning).ShouldBeTrue();
}
[Fact]
public async Task Dispose_releases_the_real_channel_cache_and_is_idempotent()
{
// Built with NO client factory, so the production DefaultClientFactory runs and a real
// GrpcChannel lands in the cache — otherwise Dispose is walking an empty dictionary and this
// test proves nothing about the thing it names. The address is loopback port 1: nothing
// listens, the connection is refused immediately, and no traffic leaves the machine.
var log = new CapturingLogger<GrpcApplyAckSender>();
var sender = new GrpcApplyAckSender(
Options.Create(new MeshTransportOptions
{
Mode = MeshTransportOptions.ModeGrpc,
CentralGrpcEndpoints = ["http://127.0.0.1:1"],
GrpcPsk = Psk,
AckTimeoutSeconds = 5,
}),
Options.Create(new AkkaClusterOptions { Roles = ["driver", "cluster-SITE-A"] }),
log);
await sender.SendAsync(Ack());
// The post must have failed (nothing is listening) — which is what put a channel in the cache.
log.Entries.Any(e => e.Level == LogLevel.Warning).ShouldBeTrue();
sender.Dispose();
Should.NotThrow(sender.Dispose);
}
// ---- fakes --------------------------------------------------------------------------------
private static AsyncUnaryCall<ReportAck> Ok() => new(
Task.FromResult(new ReportAck()),
Task.FromResult(new Metadata()),
() => Status.DefaultSuccess,
() => [],
() => { });
/// <summary>A generated client double recording the request and the call options it was given.</summary>
private sealed class RecordingClient(List<(ApplyAckMessage Request, CallOptions Options)> calls)
: CentralCommandService.CentralCommandServiceClient
{
public override AsyncUnaryCall<ReportAck> ReportApplyAsync(ApplyAckMessage request, CallOptions options)
{
calls.Add((request, options));
return Ok();
}
}
/// <summary>A generated client double whose call throws — an endpoint that is down or refuses.</summary>
private sealed class ThrowingClient(RpcException toThrow) : CentralCommandService.CentralCommandServiceClient
{
public override AsyncUnaryCall<ReportAck> ReportApplyAsync(ApplyAckMessage request, CallOptions options)
=> throw toThrow;
}
/// <summary>
/// A generated client double modelling a HUNG endpoint: it accepts the call, records it, then
/// takes real wall-clock time before failing — the shape a shared deadline cannot survive.
/// </summary>
private sealed class SlowThrowingClient(
List<(ApplyAckMessage Request, CallOptions Options)> calls, TimeSpan delay)
: CentralCommandService.CentralCommandServiceClient
{
public override AsyncUnaryCall<ReportAck> ReportApplyAsync(ApplyAckMessage request, CallOptions options)
{
calls.Add((request, options));
return new AsyncUnaryCall<ReportAck>(
HangThenFail(delay),
Task.FromResult(new Metadata()),
() => new Status(StatusCode.DeadlineExceeded, "hung"),
() => [],
() => { });
}
private static async Task<ReportAck> HangThenFail(TimeSpan delay)
{
await Task.Delay(delay);
throw new RpcException(new Status(StatusCode.DeadlineExceeded, "hung"));
}
}
/// <summary>A generated client double throwing something that is NOT an <see cref="RpcException"/>.</summary>
private sealed class FaultingClient(Exception toThrow) : CentralCommandService.CentralCommandServiceClient
{
public override AsyncUnaryCall<ReportAck> ReportApplyAsync(ApplyAckMessage request, CallOptions options)
=> throw toThrow;
}
/// <summary>Test logger that captures all log entries.</summary>
/// <typeparam name="T">The category type.</typeparam>
private sealed class CapturingLogger<T> : ILogger<T>
{
/// <summary>Gets the captured log entries.</summary>
public List<(LogLevel Level, string Message, Exception? Exception)> Entries { get; } = [];
/// <summary>Begins a scope (stub implementation).</summary>
/// <typeparam name="TState">The type of the scope state.</typeparam>
/// <param name="state">The scope state.</param>
/// <returns>A no-op scope.</returns>
public IDisposable BeginScope<TState>(TState state) where TState : notnull => NullScope.Instance;
/// <summary>Checks if logging is enabled (always true).</summary>
/// <param name="logLevel">The log level to check.</param>
/// <returns><c>true</c>.</returns>
public bool IsEnabled(LogLevel logLevel) => true;
/// <summary>Logs an entry and captures it.</summary>
/// <typeparam name="TState">The type of the log state.</typeparam>
/// <param name="logLevel">The log level.</param>
/// <param name="eventId">The event ID.</param>
/// <param name="state">The log state.</param>
/// <param name="exception">The exception, if any.</param>
/// <param name="formatter">The log message formatter.</param>
public void Log<TState>(
LogLevel logLevel, EventId eventId, TState state, Exception? exception,
Func<TState, Exception?, string> formatter)
{
Entries.Add((logLevel, formatter(state, exception), exception));
}
private sealed class NullScope : IDisposable
{
/// <summary>Gets the singleton instance.</summary>
public static NullScope Instance { get; } = new();
/// <summary>No-op.</summary>
public void Dispose() { }
}
}
}
@@ -94,8 +94,9 @@ public class NodeCommunicationActorTests : RuntimeActorTestBase
public void ApplyAck_goes_to_central_as_a_Send_not_a_SendToAll() public void ApplyAck_goes_to_central_as_a_Send_not_a_SendToAll()
{ {
// Send, not SendToAll: the deploy coordinator is one singleton behind the central comm // Send, not SendToAll: the deploy coordinator is one singleton behind the central comm
// actor's proxy. SendToAll would deliver one ack per central node and the coordinator would // actor's proxy. The coordinator's NodeId-keyed upsert absorbs a duplicate idempotently,
// count this node twice. // so the single-target rule is defence in depth and a bound on wasted work, not a
// correctness requirement.
var client = CreateTestProbe(); var client = CreateTestProbe();
var actor = Sys.ActorOf(NodeCommunicationActor.Props(client.Ref)); var actor = Sys.ActorOf(NodeCommunicationActor.Props(client.Ref));
@@ -127,4 +128,104 @@ public class NodeCommunicationActorTests : RuntimeActorTestBase
EventFilter.Warning(contains: "dropping ApplyAck").ExpectOne( EventFilter.Warning(contains: "dropping ApplyAck").ExpectOne(
TimeSpan.FromSeconds(5), () => actor.Tell(ack)); TimeSpan.FromSeconds(5), () => actor.Tell(ack));
} }
// ---- Mode=Grpc (ClusterClient→gRPC migration Task 1B.2) -------------------------------------
private static ApplyAck NewAck() => new(
new DeploymentId(Guid.NewGuid()),
new NodeId("host-a:4053"),
ApplyAckOutcome.Applied,
null,
new CorrelationId(Guid.NewGuid()));
[Fact]
public void Under_Grpc_the_ack_goes_to_the_sender_exactly_once()
{
// The gRPC counterpart of the ClusterClient.Send fact above, and it carries the same
// single-count obligation: one post per node per dispatch, because both central endpoints
// route to the SAME coordinator singleton proxy. Where the ack goes among the endpoints is the
// sender's business (GrpcApplyAckSenderTests); what this pins is that the actor hands it over
// once and does not also fan it out itself.
var sender = new RecordingAckSender();
var actor = Sys.ActorOf(NodeCommunicationActor.Props(centralClient: null, ackSender: sender));
var ack = NewAck();
actor.Tell(ack);
AwaitAssert(() => sender.Sent.Count.ShouldBe(1), TimeSpan.FromSeconds(5));
sender.Sent[0].ShouldBe(ack);
ExpectNoMsg(TimeSpan.FromMilliseconds(300));
sender.Sent.Count.ShouldBe(1);
}
[Fact]
public void A_sender_wins_over_a_cluster_client_and_the_ack_is_not_sent_twice()
{
// Defence in depth. The registration arm builds exactly one transport — under Grpc no
// ClusterClient is created at all — and if both were ever present, posting over both would
// deliver two acks for one apply. The coordinator would absorb the duplicate (NodeId-keyed
// upsert), so this pins tidiness and bounded work, not seal correctness.
var client = CreateTestProbe();
var sender = new RecordingAckSender();
var actor = Sys.ActorOf(NodeCommunicationActor.Props(client.Ref, sender));
actor.Tell(NewAck());
AwaitAssert(() => sender.Sent.Count.ShouldBe(1), TimeSpan.FromSeconds(5));
client.ExpectNoMsg(TimeSpan.FromMilliseconds(300));
}
[Fact]
public void The_ClusterClient_branch_never_touches_the_sender()
{
// Null-safe by construction: under Dps and ClusterClient the sender is not built at all, so the
// ApplyAck path must not depend on one existing.
var client = CreateTestProbe();
var actor = Sys.ActorOf(NodeCommunicationActor.Props(client.Ref));
var ack = NewAck();
actor.Tell(ack);
client.ExpectMsg<ClusterClient.Send>(TimeSpan.FromSeconds(5)).Message.ShouldBe(ack);
}
[Fact]
public void Inbound_commands_are_unaffected_by_the_presence_of_a_sender()
{
// The sender is outbound-only. A Grpc node still re-emits central's commands on its own
// EventStream exactly as every other mode does.
var subscriber = CreateTestProbe();
Sys.EventStream.Subscribe(subscriber.Ref, typeof(DispatchDeployment));
var sender = new RecordingAckSender();
var actor = Sys.ActorOf(NodeCommunicationActor.Props(centralClient: null, ackSender: sender));
var dispatch = NewDispatch();
actor.Tell(dispatch);
subscriber.ExpectMsg<DispatchDeployment>(TimeSpan.FromSeconds(5)).ShouldBe(dispatch);
sender.Sent.ShouldBeEmpty();
}
/// <summary>An <see cref="IApplyAckSender"/> double recording what the actor handed it.</summary>
private sealed class RecordingAckSender : IApplyAckSender
{
private readonly Lock _gate = new();
private readonly List<ApplyAck> _sent = [];
/// <summary>Gets a snapshot of the acks handed to this sender.</summary>
public IReadOnlyList<ApplyAck> Sent
{
get { lock (_gate) return [.. _sent]; }
}
/// <summary>Records <paramref name="ack"/>.</summary>
/// <param name="ack">The ack the actor handed over.</param>
/// <param name="cancellationToken">Unused.</param>
/// <returns>A completed task — the real sender never throws either.</returns>
public Task SendAsync(ApplyAck ack, CancellationToken cancellationToken = default)
{
lock (_gate) _sent.Add(ack);
return Task.CompletedTask;
}
}
} }
@@ -1,13 +1,17 @@
using Akka.Actor;
using Akka.Hosting; using Akka.Hosting;
using Microsoft.EntityFrameworkCore; using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting; using Microsoft.Extensions.Hosting;
using Shouldly; using Shouldly;
using Xunit; using Xunit;
using ZB.MOM.WW.OtOpcUa.Cluster;
using ZB.MOM.WW.OtOpcUa.Commons.Interfaces; using ZB.MOM.WW.OtOpcUa.Commons.Interfaces;
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Deploy;
using ZB.MOM.WW.OtOpcUa.Commons.Types; using ZB.MOM.WW.OtOpcUa.Commons.Types;
using ZB.MOM.WW.OtOpcUa.Configuration; using ZB.MOM.WW.OtOpcUa.Configuration;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions; using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Runtime.Communication;
using ZB.MOM.WW.OtOpcUa.Runtime.Historian; using ZB.MOM.WW.OtOpcUa.Runtime.Historian;
namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests; namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests;
@@ -121,6 +125,91 @@ public sealed class ServiceCollectionExtensionsTests
} }
} }
/// <summary>
/// Under <c>MeshTransport:Mode = Grpc</c> an ApplyAck produced by the driver host reaches the
/// registered <see cref="IApplyAckSender"/>, and no ClusterClient actor exists
/// (ClusterClient→gRPC migration Task 1B.2).
/// </summary>
/// <remarks>
/// This is the fact that pins the <c>ackRouter</c> gate in <c>WithOtOpcUaRuntimeActors</c>
/// (<c>useClusterClient || useGrpc ? nodeComm : null</c>). Narrowed back to ClusterClient-only
/// that gate still compiles and every unit test still passes, while every ack on a Grpc node
/// goes out on the <c>deployment-acks</c> DPS topic instead — where, in the split-mesh topology
/// Grpc exists to serve, central is not subscribed. The whole transport would be dead and
/// nothing below this level of the stack could tell.
///
/// The dispatch names a deployment that does not exist in the in-memory ConfigDb, so the apply
/// fails and the host emits a Failed ack — deterministic, synchronous, and it exercises the
/// real <c>SendAck</c> path rather than poking the router directly.
/// </remarks>
[Fact]
public async Task Under_Grpc_a_driver_host_ack_reaches_the_ack_sender_and_no_ClusterClient_exists()
{
var sender = new RecordingAckSender();
using var host = BuildRuntimeActorHost(services =>
{
services.AddSingleton<IApplyAckSender>(sender);
services.Configure<MeshTransportOptions>(o =>
{
o.Mode = MeshTransportOptions.ModeGrpc;
o.CentralGrpcEndpoints = ["http://central-1:4055"];
o.GrpcPsk = "rig-key";
});
});
await host.StartAsync();
try
{
var system = host.Services.GetRequiredService<ActorSystem>();
var driverHost = host.Services.GetRequiredService<IRequiredActor<DriverHostActorKey>>();
driverHost.ActorRef.Tell(new DispatchDeployment(
new DeploymentId(Guid.NewGuid()),
new RevisionHash("rev-that-does-not-exist"),
new CorrelationId(Guid.NewGuid())));
// The ack travelled DriverHostActor -> NodeCommunicationActor -> IApplyAckSender. Under a
// ClusterClient-only gate it would have gone to DPS and this never completes.
var deadline = DateTime.UtcNow.AddSeconds(15);
while (sender.Sent.Count == 0 && DateTime.UtcNow < deadline)
await Task.Delay(50);
sender.Sent.Count.ShouldBe(1);
// And the Akka cross-mesh association is genuinely gone, not merely unused.
await Should.ThrowAsync<ActorNotFoundException>(() => system
.ActorSelection($"/user/{ServiceCollectionExtensions.CentralClusterClientName}")
.ResolveOne(TimeSpan.FromMilliseconds(500)));
}
finally
{
await host.StopAsync();
}
}
/// <summary>An <see cref="IApplyAckSender"/> double recording what the node comm actor handed it.</summary>
private sealed class RecordingAckSender : IApplyAckSender
{
private readonly Lock _gate = new();
private readonly List<ApplyAck> _sent = [];
/// <summary>Gets a snapshot of the acks posted through this sender.</summary>
public IReadOnlyList<ApplyAck> Sent
{
get { lock (_gate) return [.. _sent]; }
}
/// <summary>Records <paramref name="ack"/>.</summary>
/// <param name="ack">The ack handed over.</param>
/// <param name="cancellationToken">Unused.</param>
/// <returns>A completed task.</returns>
public Task SendAsync(ApplyAck ack, CancellationToken cancellationToken = default)
{
lock (_gate) _sent.Add(ack);
return Task.CompletedTask;
}
}
/// <summary>Builds a driver-role host that runs <c>WithOtOpcUaRuntimeActors</c>, with optional /// <summary>Builds a driver-role host that runs <c>WithOtOpcUaRuntimeActors</c>, with optional
/// extra DI registrations applied before <c>AddAkka</c>.</summary> /// extra DI registrations applied before <c>AddAkka</c>.</summary>
private static IHost BuildRuntimeActorHost(Action<IServiceCollection>? extra) private static IHost BuildRuntimeActorHost(Action<IServiceCollection>? extra)