diff --git a/docs/plans/2026-08-06-clusterclient-to-grpc-live-gate.md b/docs/plans/2026-08-06-clusterclient-to-grpc-live-gate.md index e18b860b..64f68181 100644 --- a/docs/plans/2026-08-06-clusterclient-to-grpc-live-gate.md +++ b/docs/plans/2026-08-06-clusterclient-to-grpc-live-gate.md @@ -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: -site-a-1/-2 → `Grpc`; site-b stays `ClusterClient`; central back on `ClusterClient` mode +Branch `feat/grpc-central-ack` @ `4a2a3d3a` (rebased on `master`, which carries the merged 1A). +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. | # | 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 | -| 2 | Site-b acks still arrive via ClusterClient in the same deployment — per-pair coexistence proven | PENDING | -| 3 | Exactly **one** ack row per site-a node per dispatch (no double-count from endpoint failover) | PENDING | -| 4 | Kill central-1 before the apply completes: site-a acks fail over to central-2's endpoint; deployment still seals | PENDING | -| 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 | -| 6 | ConfigServe (`DeploymentArtifactService`) on the same listener unaffected: fetches succeed, 0 auth failures | 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 | **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) | **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 (`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** | +| **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` | +| 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://: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 ' 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 = ' | 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 = '' … +$ 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-` 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://: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 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:** ` **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: `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 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`**; -soak ≥ 2 h with 0 `PermissionDenied` noise — PSK drift detector). **Phase 2 DoD:** every -check PASS in the live-gate doc; rig default committed; tracking updated. **STOP.** +soak ≥ 2 h with 0 `PermissionDenied` noise — PSK drift detector; the soak error-grep must be +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: `rg -i "clusterclient|receptionist" src tests docker-dev docs` → only historical 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 @@ -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, `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. -**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; (6) **boundary check: no Akka association between any site container and central** (was the baseline gap); (7) restart discipline: full-rig restart clean, **zero -receptionist/ClusterClient log lines anywhere**; (8) `MeshGrpcBoundaryTests` + full suite -green on the merge SHA. Close any carry-forwards named by earlier gates. Tracking doc final -row; program CLOSED. +receptionist/ClusterClient log lines anywhere**; (8) `MeshGrpcBoundaryTests` (targeted) + +the program's TERMINAL full-suite run, green on the merge SHA — this is Phase 4's one full +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. --- diff --git a/docs/plans/2026-08-06-clusterclient-to-grpc.md.tasks.json b/docs/plans/2026-08-06-clusterclient-to-grpc.md.tasks.json index f218add7..f5e13fca 100644 --- a/docs/plans/2026-08-06-clusterclient-to-grpc.md.tasks.json +++ b/docs/plans/2026-08-06-clusterclient-to-grpc.md.tasks.json @@ -149,10 +149,12 @@ "parallelizableWith": [ "1A.1 (other worktree)" ], - "status": "pending", + "status": "completed", "blockedBy": [ 5 - ] + ], + "commit": "14dd07d2 (rebased 7e7df337)", + "review": "spec ✅ + code ✅" }, { "id": 10, @@ -163,10 +165,12 @@ "parallelizableWith": [ "1A.2 (other worktree)" ], - "status": "pending", + "status": "completed", "blockedBy": [ 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, @@ -175,11 +179,13 @@ "branch": "feat/grpc-central-ack", "classification": "small", "parallelizableWith": [], - "status": "pending", + "status": "completed", "blockedBy": [ 8, 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, @@ -191,8 +197,10 @@ "status": "pending", "blockedBy": [ 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, @@ -229,6 +237,20 @@ "blockedBy": [ 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" diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Cluster/MeshTransportOptionsValidator.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Cluster/MeshTransportOptionsValidator.cs index ad47fc88..5641672e 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Cluster/MeshTransportOptionsValidator.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Cluster/MeshTransportOptionsValidator.cs @@ -94,9 +94,10 @@ public sealed class MeshTransportOptionsValidator : IValidateOptions()); var isDriver = Array.IndexOf(roles, RoleParser.Driver) >= 0; + var isAdmin = Array.IndexOf(roles, RoleParser.Admin) >= 0; if (isGrpc && isDriver) { @@ -166,6 +168,37 @@ public sealed class MeshTransportOptionsValidator : IValidateOptions 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 ? ValidateOptionsResult.Success : ValidateOptionsResult.Fail(string.Join(" ", errors)); diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Grpc/CentralCommandGrpcService.cs b/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Grpc/CentralCommandGrpcService.cs new file mode 100644 index 00000000..5fe5ea78 --- /dev/null +++ b/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Grpc/CentralCommandGrpcService.cs @@ -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; + +/// +/// Central-side ingress for a node's deployment acknowledgment (ClusterClient→gRPC migration +/// Phase 1B). Decodes an and hands it to the node-local +/// CentralCommunicationActor — 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 ConfigServe:GrpcListenPort > 0. +/// +/// +/// +/// The actor is resolved per call, not in the constructor. Kestrel answers before +/// Akka.Hosting has finished spawning the actor graph, so a ctor lookup would throw +/// during the boot window and surface as an opaque Unknown. Resolving per call turns +/// that window into — 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. +/// +/// +/// Decoding is graded before the registry. A message that cannot decode is +/// 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 ) — 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 Applied and a guessed ack seals a +/// deployment nobody applied. +/// +/// +/// The rejection log carries the wire correlation id verbatim. 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. +/// +/// +/// One-way semantics, preserved. Tell, not Ask: the returned +/// 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. +/// +/// +public sealed class CentralCommandGrpcService : GeneratedServiceBase +{ + private readonly ActorRegistry _registry; + private readonly ILogger _log; + + /// Initializes a new instance of the class. + /// The actor registry holding the node's CentralCommunicationActor. + /// Logger. + public CentralCommandGrpcService(ActorRegistry registry, ILogger log) + { + _registry = registry; + _log = log; + } + + /// + public override Task 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(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()); + } +} diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Program.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Program.cs index 2fb407a3..80f3048b 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Program.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Program.cs @@ -744,6 +744,16 @@ if (hasDriver && syncListenPort > 0) if (hasAdmin && configServeGrpcPort > 0) { app.MapGrpcService(); + + // 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(); } // Driver-node live-telemetry stream (per-cluster mesh Phase 5). Central dials in; gated by diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Communication/GrpcApplyAckSender.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Communication/GrpcApplyAckSender.cs new file mode 100644 index 00000000..b47b2a96 --- /dev/null +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Communication/GrpcApplyAckSender.cs @@ -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; + +/// +/// Posts this node's to central's CentralCommandService as a unary +/// gRPC call (ClusterClient→gRPC migration Task 1B.2), the MeshTransport:Mode = Grpc +/// replacement for the ClusterClient Send. +/// +/// +/// +/// One ordered pass, stopping at the first success. +/// is an availability list, not a +/// fan-out list — the semantic of the ClusterClient.Send being replaced (Send, +/// deliberately not SendToAll). +/// +/// +/// Delivery is at-most-once per SUCCESSFUL attempt, but the walk as a whole is +/// at-least-once, 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. +/// ConfigPublishCoordinator.HandleAck absorbs that idempotently: acks land in a +/// NodeId-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. +/// +/// +/// No retry beyond that single pass, and no queue. 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. +/// +/// +/// All endpoints failing is a DROP — but a loud one. 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". +/// +/// +/// The bearer never reaches a log. 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 +/// would write the cluster's pre-shared key into the node log of every +/// deployment that could not reach central. +/// +/// +/// The cluster id comes from this node's own cluster-<id> Akka role, read +/// through off the same bound Cluster:Roles that +/// ClusterRoleInfo reads (the OTOPCUA_ROLES fallback is already applied to those +/// options by AddOtOpcUaCluster's PostConfigure, 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. +/// +/// +public sealed class GrpcApplyAckSender : IApplyAckSender, IDisposable +{ + /// The cluster the caller claims to be; central grades the bearer against ITS key. + private const string ClusterHeader = "x-otopcua-cluster"; + + private readonly MeshTransportOptions _options; + private readonly string? _clusterId; + private readonly ILogger _log; + private readonly Func _clientFactory; + private readonly ConcurrentDictionary _channels = new(StringComparer.Ordinal); + + /// Initializes a new instance of the class. + /// Mesh-transport options; supplies the endpoints, the PSK and the deadline. + /// + /// Cluster options, read only for this node's roles — the cluster-<id> one names + /// the cluster this node authenticates as. + /// + /// Logger. + /// + /// Seam for tests: builds the generated client for an endpoint. Defaults to a real + /// (one cached per endpoint). + /// + public GrpcApplyAckSender( + IOptions options, + IOptions clusterOptions, + ILogger log, + Func? 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}' 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); + } + } + + /// + 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(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); + } + + /// + public void Dispose() + { + foreach (var channel in _channels.Values) + channel.Dispose(); + _channels.Clear(); + } +} diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Communication/IApplyAckSender.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Communication/IApplyAckSender.cs new file mode 100644 index 00000000..092d30ab --- /dev/null +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Communication/IApplyAckSender.cs @@ -0,0 +1,40 @@ +using ZB.MOM.WW.OtOpcUa.Commons.Messages.Deploy; + +namespace ZB.MOM.WW.OtOpcUa.Runtime.Communication; + +/// +/// Posts this node's deployment acknowledgment to central over the mesh_transport.v1 +/// command plane (ClusterClient→gRPC migration Phase 1B). The seam +/// uses under MeshTransport:Mode = Grpc, in place of +/// the ClusterClient.Send it uses under ClusterClient. +/// +/// +/// +/// One ack per node per dispatch. Whatever the implementation does about redundancy, it +/// posts to one central endpoint at a time rather than fanning out — the semantic of the +/// ClusterClient.Send this replaces (Send, deliberately not SendToAll). +/// This is defence in depth, not a correctness dependency: the coordinator counts acks +/// in a NodeId-keyed dictionary behind an expected-set membership guard +/// (ConfigPublishCoordinator.HandleAck), 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. +/// +/// +/// Fire-and-forget, and it never throws. 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. +/// +/// +public interface IApplyAckSender +{ + /// Posts to central, or logs why it could not be. + /// The per-node apply acknowledgment. + /// Cancels the in-flight call. + /// + /// A task that completes when the attempt finishes, successfully or not. It never faults — + /// see the remarks. + /// + Task SendAsync(ApplyAck ack, CancellationToken cancellationToken = default); +} diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Communication/NodeCommunicationActor.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Communication/NodeCommunicationActor.cs index fe192b2d..88702dc8 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Communication/NodeCommunicationActor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Communication/NodeCommunicationActor.cs @@ -37,23 +37,31 @@ namespace ZB.MOM.WW.OtOpcUa.Runtime.Communication; public sealed class NodeCommunicationActor : ReceiveActor { private readonly IActorRef? _centralClient; + private readonly IApplyAckSender? _ackSender; private readonly ILoggingAdapter _log = Context.GetLogger(); /// Creates the props for this actor. /// /// ClusterClient dialling central's receptionist, or under /// MeshTransport:Mode = Dps, where acks still go out over DistributedPubSub from - /// DriverHostActor and this actor never sees them. + /// DriverHostActor and this actor never sees them — and under + /// Mode = Grpc, where no ClusterClient is created at all. + /// + /// + /// The gRPC ack transport under MeshTransport:Mode = Grpc, or + /// under Dps and ClusterClient. /// /// The props. - public static Props Props(IActorRef? centralClient) => - Akka.Actor.Props.Create(() => new NodeCommunicationActor(centralClient)); + public static Props Props(IActorRef? centralClient, IApplyAckSender? ackSender = null) => + Akka.Actor.Props.Create(() => new NodeCommunicationActor(centralClient, ackSender)); /// Initializes a new instance of the class. /// ClusterClient dialling central, or . - public NodeCommunicationActor(IActorRef? centralClient) + /// The gRPC ack transport, or . + public NodeCommunicationActor(IActorRef? centralClient, IApplyAckSender? ackSender = null) { _centralClient = centralClient; + _ackSender = ackSender; // 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 @@ -75,20 +83,54 @@ public sealed class NodeCommunicationActor : ReceiveActor 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) { _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 " + "naming this node even though it applied successfully. Check " - + "MeshTransport:CentralContactPoints", + + "MeshTransport:CentralContactPoints (ClusterClient mode) or " + + "MeshTransport:CentralGrpcEndpoints (Grpc mode)", ack.DeploymentId); return; } // 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 - // coordinator would count the same node twice. + // actor forwards to its proxy. Its NodeId-keyed upsert absorbs a duplicate idempotently, + // 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)); } } diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs index 255f2590..7d14508e 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs @@ -374,8 +374,9 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers /// Database context factory for configuration database access. /// The local cluster node identifier. /// Where ApplyAcks are sent. A direct coordinator handle in tests; under - /// MeshTransport:Mode = ClusterClient the node's NodeCommunicationActor, which relays - /// across the boundary. Null publishes on the deployment-acks DPS topic. + /// MeshTransport:Mode = ClusterClient and = Grpc the node's + /// NodeCommunicationActor, which relays across the boundary over whichever transport it was + /// built with. Null publishes on the deployment-acks DPS topic. /// Optional driver factory; defaults to null factory if not provided. /// Optional set of roles assigned to the local node. /// Optional actor reference for dependency multiplexing. @@ -458,8 +459,9 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers /// Database context factory for configuration database access. /// The local cluster node identifier. /// Where ApplyAcks are sent. A direct coordinator handle in tests; under - /// MeshTransport:Mode = ClusterClient the node's NodeCommunicationActor, which relays - /// across the boundary. Null publishes on the deployment-acks DPS topic. + /// MeshTransport:Mode = ClusterClient and = Grpc the node's + /// NodeCommunicationActor, which relays across the boundary over whichever transport it was + /// built with. Null publishes on the deployment-acks DPS topic. /// Optional driver factory; defaults to null factory if not provided. /// Optional set of roles assigned to the local node. /// Optional actor reference for dependency multiplexing. @@ -2490,11 +2492,11 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers var ack = new ApplyAck(deploymentId, _localNode, outcome, failureReason, correlation); if (_ackRouter is not null) { - // Either a direct coordinator handle (tests) or, under - // MeshTransport:Mode = ClusterClient, the NodeCommunicationActor that relays this - // across the boundary. Renamed from _coordinatorOverride in per-cluster mesh Phase 2: - // under ClusterClient mode this ref is emphatically NOT a coordinator, and the old - // name would send the next reader looking for one. + // Either a direct coordinator handle (tests) or, under MeshTransport:Mode = ClusterClient + // and = Grpc alike, the NodeCommunicationActor that relays this across the boundary (it + // owns the choice of transport, not this actor). Renamed from _coordinatorOverride in + // per-cluster mesh Phase 2: under those two modes this ref is emphatically NOT a + // coordinator, and the old name would send the next reader looking for one. _ackRouter.Tell(ack); } else diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ServiceCollectionExtensions.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ServiceCollectionExtensions.cs index eeda4156..dd7c05ff 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ServiceCollectionExtensions.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ServiceCollectionExtensions.cs @@ -1,6 +1,7 @@ using System.Collections.Immutable; using Akka.Actor; using Akka.Cluster.Tools.Client; +using Akka.Event; using Akka.Hosting; using Microsoft.EntityFrameworkCore; 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 // with no gate at all, and both nodes would write every sample twice with nothing to say so. services.TryAddSingleton(sp => sp.GetRequiredService()); + + // 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(sp => new GrpcApplyAckSender( + sp.GetRequiredService>(), + sp.GetRequiredService>(), + sp.GetService()?.CreateLogger() + ?? NullLogger.Instance)); return services; } @@ -389,6 +410,8 @@ public static class ServiceCollectionExtensions var meshOptions = resolver.GetService>().Value; var useClusterClient = string.Equals( meshOptions.Mode, MeshTransportOptions.ModeClusterClient, StringComparison.OrdinalIgnoreCase); + var useGrpc = string.Equals( + meshOptions.Mode, MeshTransportOptions.ModeGrpc, StringComparison.OrdinalIgnoreCase); IActorRef? centralClient = null; if (useClusterClient) @@ -406,12 +429,33 @@ public static class ServiceCollectionExtensions 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(); + 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( - NodeCommunicationActor.Props(centralClient), + NodeCommunicationActor.Props(centralClient, ackSender), MeshPaths.NodeCommunicationName); - // Registered with the receptionist in BOTH modes: under Dps nothing dials it, and an - // idle registration costs nothing — whereas registering only under ClusterClient would + // Registered with the receptionist in ALL modes: under Dps and Grpc nothing dials it, and + // an idle registration costs nothing — whereas registering only under ClusterClient would // make flipping the flag require a node restart before central could reach it. ClusterClientReceptionist.Get(system).RegisterService(nodeComm); registry.Register(nodeComm); @@ -523,9 +567,12 @@ public static class ServiceCollectionExtensions : null; var driverHost = system.ActorOf( - // ackRouter: under ClusterClient mode the node comm actor relays ApplyAcks across - // the boundary; under Dps it stays null and the host publishes on the - // deployment-acks topic exactly as before. + // ackRouter: under ClusterClient AND Grpc the node comm actor relays ApplyAcks across + // the boundary (which transport it then uses is its own business — the ClusterClient + // 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 // gated on hasAdmin and driver-only nodes are forced to FetchAndCache. DriverHostActor // 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 // host when no store was wired). DriverHostActor.Props(dbFactory, roleInfo.LocalNode, - ackRouter: useClusterClient ? nodeComm : null, + ackRouter: useClusterClient || useGrpc ? nodeComm : null, driverFactory: driverFactory, localRoles: roleInfo.LocalRoles, dependencyMux: mux, opcUaPublishActor: publishActor, diff --git a/tests/Core/ZB.MOM.WW.OtOpcUa.Cluster.Tests/MeshTransportOptionsValidatorTests.cs b/tests/Core/ZB.MOM.WW.OtOpcUa.Cluster.Tests/MeshTransportOptionsValidatorTests.cs index 1fb4ce46..294897b2 100644 --- a/tests/Core/ZB.MOM.WW.OtOpcUa.Cluster.Tests/MeshTransportOptionsValidatorTests.cs +++ b/tests/Core/ZB.MOM.WW.OtOpcUa.Cluster.Tests/MeshTransportOptionsValidatorTests.cs @@ -20,9 +20,18 @@ namespace ZB.MOM.WW.OtOpcUa.Cluster.Tests; [Collection(RoleEnvironmentCollection.Name)] 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(), roles); + + /// + /// As , plus arbitrary configuration + /// keys — the admin-side rule cross-reads ConfigServe:GrpcListenPort, which lives + /// outside this options class entirely. + /// + private static ValidateOptionsResult Validate( + MeshTransportOptions o, Dictionary extra, params string[] roles) { - var pairs = new Dictionary(); + var pairs = new Dictionary(extra); for (var i = 0; i < roles.Length; i++) { pairs[$"Cluster:Roles:{i}"] = roles[i]; @@ -34,6 +43,9 @@ public class MeshTransportOptionsValidatorTests .Validate(MeshTransportOptions.SectionName, o); } + private static Dictionary ConfigServePort(string? value) => + new() { ["ConfigServe:GrpcListenPort"] = value }; + private static MeshTransportOptions WellFormedGrpc() => new() { 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 // 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. + // (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( new MeshTransportOptions { Mode = MeshTransportOptions.ModeGrpc }, + ConfigServePort("4055"), "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] 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(); } + [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] public void ClusterPsks_defaults_to_an_empty_case_insensitive_map() { diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.AdminUI.Tests/Grpc/CentralCommandGrpcServiceTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.AdminUI.Tests/Grpc/CentralCommandGrpcServiceTests.cs new file mode 100644 index 00000000..b2242985 --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.AdminUI.Tests/Grpc/CentralCommandGrpcServiceTests.cs @@ -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; + +/// +/// 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 +/// CentralCommunicationActor — the same actor the ClusterClient receptionist delivered to, +/// whose forward-to-the-coordinator path is untouched. +/// +/// +/// 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 so the caller +/// retries instead of recording a permanent failure against a node that is fine. +/// +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.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(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(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(probe.Ref); + var service = Service(registry); + + await service.ReportApply(WellFormed(), new FakeServerCallContext()); + probe.ExpectMsg(TimeSpan.FromSeconds(3)).FailureReason.ShouldBeNull(); + + var blank = WellFormed(); + blank.FailureReason = ""; + await service.ReportApply(blank, new FakeServerCallContext()); + probe.ExpectMsg(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(() => + 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(probe.Ref); + + var request = WellFormed(); + request.Outcome = ApplyOutcome.Unspecified; + + var ex = await Should.ThrowAsync(() => + 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(probe.Ref); + + var ex = await Should.ThrowAsync(() => + 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(probe.Ref); + + var ex = await Should.ThrowAsync(() => + 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(() => + 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(); + var request = WellFormed(); + request.CorrelationId = "correlation-that-never-parses"; + request.DeploymentId = "not-a-guid"; + + await Should.ThrowAsync(() => + 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(); + var request = WellFormed(); + + await Should.ThrowAsync(() => + 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); + } + + /// Test logger that captures all log entries. + /// The category type. + private sealed class CapturingLogger : ILogger + { + /// Gets the captured log entries. + public List<(LogLevel Level, string Message, Exception? Exception)> Entries { get; } = []; + + /// Begins a scope (stub implementation). + /// The type of the scope state. + /// The scope state. + /// A no-op scope. + public IDisposable BeginScope(TState state) where TState : notnull => NullScope.Instance; + + /// Checks if logging is enabled (always true). + /// The log level to check. + /// true. + public bool IsEnabled(LogLevel logLevel) => true; + + /// Logs an entry and captures it. + /// The type of the log state. + /// The log level. + /// The event ID. + /// The log state. + /// The exception, if any. + /// The log message formatter. + public void Log( + LogLevel logLevel, EventId eventId, TState state, Exception? exception, + Func formatter) + { + Entries.Add((logLevel, formatter(state, exception), exception)); + } + + private sealed class NullScope : IDisposable + { + /// Gets the singleton instance. + public static NullScope Instance { get; } = new(); + + /// No-op. + public void Dispose() { } + } + } + + /// Minimal — the service reads only the method path. + 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>()); + + protected override ContextPropagationToken CreatePropagationTokenCore(ContextPropagationOptions? options) => + throw new NotSupportedException(); + + protected override Task WriteResponseHeadersAsyncCore(Metadata responseHeaders) => Task.CompletedTask; + } +} diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Communication/GrpcApplyAckSenderTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Communication/GrpcApplyAckSenderTests.cs new file mode 100644 index 00000000..bd6603f8 --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Communication/GrpcApplyAckSenderTests.cs @@ -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; + +/// +/// 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 +/// CentralGrpcEndpoints, the per-call credentials, and the drop that used to be silent. +/// +/// +/// +/// The load-bearing fact is that ONE endpoint is dialled on success. This replaces an +/// Akka ClusterClient.Send, which is single-target at-most-once — not a +/// SendToAll. Both central nodes route an ack to the same +/// ConfigPublishCoordinator 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. +/// +/// +/// 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. +/// +/// +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 factory, + ILogger? 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.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(); + 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(); + 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(); + 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(); + 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(); + 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(); + 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(); + 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(); + 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(); + 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 Ok() => new( + Task.FromResult(new ReportAck()), + Task.FromResult(new Metadata()), + () => Status.DefaultSuccess, + () => [], + () => { }); + + /// A generated client double recording the request and the call options it was given. + private sealed class RecordingClient(List<(ApplyAckMessage Request, CallOptions Options)> calls) + : CentralCommandService.CentralCommandServiceClient + { + public override AsyncUnaryCall ReportApplyAsync(ApplyAckMessage request, CallOptions options) + { + calls.Add((request, options)); + return Ok(); + } + } + + /// A generated client double whose call throws — an endpoint that is down or refuses. + private sealed class ThrowingClient(RpcException toThrow) : CentralCommandService.CentralCommandServiceClient + { + public override AsyncUnaryCall ReportApplyAsync(ApplyAckMessage request, CallOptions options) + => throw toThrow; + } + + /// + /// 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. + /// + private sealed class SlowThrowingClient( + List<(ApplyAckMessage Request, CallOptions Options)> calls, TimeSpan delay) + : CentralCommandService.CentralCommandServiceClient + { + public override AsyncUnaryCall ReportApplyAsync(ApplyAckMessage request, CallOptions options) + { + calls.Add((request, options)); + return new AsyncUnaryCall( + HangThenFail(delay), + Task.FromResult(new Metadata()), + () => new Status(StatusCode.DeadlineExceeded, "hung"), + () => [], + () => { }); + } + + private static async Task HangThenFail(TimeSpan delay) + { + await Task.Delay(delay); + throw new RpcException(new Status(StatusCode.DeadlineExceeded, "hung")); + } + } + + /// A generated client double throwing something that is NOT an . + private sealed class FaultingClient(Exception toThrow) : CentralCommandService.CentralCommandServiceClient + { + public override AsyncUnaryCall ReportApplyAsync(ApplyAckMessage request, CallOptions options) + => throw toThrow; + } + + /// Test logger that captures all log entries. + /// The category type. + private sealed class CapturingLogger : ILogger + { + /// Gets the captured log entries. + public List<(LogLevel Level, string Message, Exception? Exception)> Entries { get; } = []; + + /// Begins a scope (stub implementation). + /// The type of the scope state. + /// The scope state. + /// A no-op scope. + public IDisposable BeginScope(TState state) where TState : notnull => NullScope.Instance; + + /// Checks if logging is enabled (always true). + /// The log level to check. + /// true. + public bool IsEnabled(LogLevel logLevel) => true; + + /// Logs an entry and captures it. + /// The type of the log state. + /// The log level. + /// The event ID. + /// The log state. + /// The exception, if any. + /// The log message formatter. + public void Log( + LogLevel logLevel, EventId eventId, TState state, Exception? exception, + Func formatter) + { + Entries.Add((logLevel, formatter(state, exception), exception)); + } + + private sealed class NullScope : IDisposable + { + /// Gets the singleton instance. + public static NullScope Instance { get; } = new(); + + /// No-op. + public void Dispose() { } + } + } +} diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Communication/NodeCommunicationActorTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Communication/NodeCommunicationActorTests.cs index 7cc06aca..473d2d02 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Communication/NodeCommunicationActorTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Communication/NodeCommunicationActorTests.cs @@ -94,8 +94,9 @@ public class NodeCommunicationActorTests : RuntimeActorTestBase 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 - // actor's proxy. SendToAll would deliver one ack per central node and the coordinator would - // count this node twice. + // actor's proxy. The coordinator's NodeId-keyed upsert absorbs a duplicate idempotently, + // so the single-target rule is defence in depth and a bound on wasted work, not a + // correctness requirement. var client = CreateTestProbe(); var actor = Sys.ActorOf(NodeCommunicationActor.Props(client.Ref)); @@ -127,4 +128,104 @@ public class NodeCommunicationActorTests : RuntimeActorTestBase EventFilter.Warning(contains: "dropping ApplyAck").ExpectOne( 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(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(TimeSpan.FromSeconds(5)).ShouldBe(dispatch); + sender.Sent.ShouldBeEmpty(); + } + + /// An double recording what the actor handed it. + private sealed class RecordingAckSender : IApplyAckSender + { + private readonly Lock _gate = new(); + private readonly List _sent = []; + + /// Gets a snapshot of the acks handed to this sender. + public IReadOnlyList Sent + { + get { lock (_gate) return [.. _sent]; } + } + + /// Records . + /// The ack the actor handed over. + /// Unused. + /// A completed task — the real sender never throws either. + public Task SendAsync(ApplyAck ack, CancellationToken cancellationToken = default) + { + lock (_gate) _sent.Add(ack); + return Task.CompletedTask; + } + } } diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/ServiceCollectionExtensionsTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/ServiceCollectionExtensionsTests.cs index 986bcc76..f649d30d 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/ServiceCollectionExtensionsTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/ServiceCollectionExtensionsTests.cs @@ -1,13 +1,17 @@ +using Akka.Actor; using Akka.Hosting; using Microsoft.EntityFrameworkCore; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Hosting; using Shouldly; using Xunit; +using ZB.MOM.WW.OtOpcUa.Cluster; 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.Configuration; using ZB.MOM.WW.OtOpcUa.Core.Abstractions; +using ZB.MOM.WW.OtOpcUa.Runtime.Communication; using ZB.MOM.WW.OtOpcUa.Runtime.Historian; namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests; @@ -121,6 +125,91 @@ public sealed class ServiceCollectionExtensionsTests } } + /// + /// Under MeshTransport:Mode = Grpc an ApplyAck produced by the driver host reaches the + /// registered , and no ClusterClient actor exists + /// (ClusterClient→gRPC migration Task 1B.2). + /// + /// + /// This is the fact that pins the ackRouter gate in WithOtOpcUaRuntimeActors + /// (useClusterClient || useGrpc ? nodeComm : null). 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 deployment-acks 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 SendAck path rather than poking the router directly. + /// + [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(sender); + services.Configure(o => + { + o.Mode = MeshTransportOptions.ModeGrpc; + o.CentralGrpcEndpoints = ["http://central-1:4055"]; + o.GrpcPsk = "rig-key"; + }); + }); + + await host.StartAsync(); + try + { + var system = host.Services.GetRequiredService(); + var driverHost = host.Services.GetRequiredService>(); + + 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(() => system + .ActorSelection($"/user/{ServiceCollectionExtensions.CentralClusterClientName}") + .ResolveOne(TimeSpan.FromMilliseconds(500))); + } + finally + { + await host.StopAsync(); + } + } + + /// An double recording what the node comm actor handed it. + private sealed class RecordingAckSender : IApplyAckSender + { + private readonly Lock _gate = new(); + private readonly List _sent = []; + + /// Gets a snapshot of the acks posted through this sender. + public IReadOnlyList Sent + { + get { lock (_gate) return [.. _sent]; } + } + + /// Records . + /// The ack handed over. + /// Unused. + /// A completed task. + public Task SendAsync(ApplyAck ack, CancellationToken cancellationToken = default) + { + lock (_gate) _sent.Add(ack); + return Task.CompletedTask; + } + } + /// Builds a driver-role host that runs WithOtOpcUaRuntimeActors, with optional /// extra DI registrations applied before AddAkka. private static IHost BuildRuntimeActorHost(Action? extra)