fix(redundancy): a failed OPC UA endpoint drops ServiceLevel to 0 and steps the Primary down (lmxopcua#762)
v2-ci / unit-tests (push) Has been cancelled

IOpcUaEndpointState (review-remediation lane C3) recorded whether the SDK started, but only the
health check read it: a fused node whose endpoint failed (port 4840 taken, ...) still advertised
250/240 and stayed driver Primary, so the Primary-gated data plane ran on a node no client could
reach while the reachable peer refused writes as Secondary.

- Move IOpcUaEndpointState/OpcUaEndpointState/OpcUaEndpointStatus from Host to Commons.OpcUa so
  the Runtime can consume it.
- ServiceLevelCalculator: new NodeHealthInputs.EndpointListening (default true). false -> 0,
  ahead of every other input including the +10 Primary bonus.
- OpcUaPublishActor: takes the endpoint state, feeds EndpointListening (Failed only; NotStarted /
  Stopped are boot/shutdown windows) on the DB-backed and DB-less paths, and the legacy role-only
  seam returns 0 for a failed endpoint too.
- New OpcUaEndpointStepDownActor (one per driver node): a Primary (ClusterActiveNode.SelfIsActive,
  the election's own primitive) with a Failed endpoint gracefully Leaves its mesh - the manual
  failover mechanism - so the peer becomes the oldest Up driver member, hosts the singletons and is
  Primary. The election itself is not bent. Guarded on a peer having recently reported its own
  endpoint listening (EndpointReport on the new opcua-endpoint-state DPS topic), so a pair with
  both endpoints failed does not bounce the role and restart nodes forever, and a lone node never
  shuts itself down.
- Runtime now references ZB.MOM.WW.Health.Akka (ClusterActiveNode).
- Docs: Redundancy.md tier table + new "Endpoint start failure: Primary step-down" section;
  cluster-topology rule; HealthEndpoints / OpcUaEndpointHealthCheck comments (F-C1 landed).

Tests: ServiceLevelCalculatorTests (+5), OpcUaPublishActorEndpointStateTests (7: DB-less,
DB-backed, legacy seam, NotStarted/Stopped do not demote, a later failure demotes on the next
tick), OpcUaEndpointStepDownActorTests (3, real two-node mesh: holds while the peer is not
listening then leaves and the peer becomes Primary; failed Secondary and healthy Primary never
leave). Each was mutation-checked: removing the calculator/legacy guard, the Leave, the
peer-listening guard or the Primary guard fails the corresponding tests.
This commit is contained in:
Joseph Doherty
2026-11-09 12:15:59 -05:00
parent 8ef5adc63a
commit acfb4fbaca
13 changed files with 668 additions and 30 deletions
+2
View File
@@ -26,6 +26,8 @@ The server supports non-transparent warm/hot redundancy via the `Redundancy` sec
- **Downing is `auto-down`, not `keep-oldest`.** `Cluster:SplitBrainResolverStrategy` (default `auto-down`) selects the provider via `ServiceCollectionExtensions.BuildDowningHocon`; an unknown value fails host start. A two-node pair running the previous `keep-oldest` + `down-if-alone` **could not survive a crash of the oldest node**: Akka's `KeepOldest.OldestDecision` only lets `down-if-alone` rescue a side holding ≥ 2 members, so the 1-vs-1 survivor downs *itself* and exits. `down-if-alone` is a 3+-node feature. The accepted trade for `auto-down` is dual-active during a genuine network partition. **Its live gate is CLOSED** — Phase 7 drill D4 (2026-07-24) ran the 1-vs-1 crash-the-oldest survival drill on the three-mesh docker-dev rig and every survivor stayed Up, never self-downing (`c50ebcf7`; `docs/plans/2026-07-24-mesh-phase7-failover-drills.md`). - **Downing is `auto-down`, not `keep-oldest`.** `Cluster:SplitBrainResolverStrategy` (default `auto-down`) selects the provider via `ServiceCollectionExtensions.BuildDowningHocon`; an unknown value fails host start. A two-node pair running the previous `keep-oldest` + `down-if-alone` **could not survive a crash of the oldest node**: Akka's `KeepOldest.OldestDecision` only lets `down-if-alone` rescue a side holding ≥ 2 members, so the 1-vs-1 survivor downs *itself* and exits. `down-if-alone` is a 3+-node feature. The accepted trade for `auto-down` is dual-active during a genuine network partition. **Its live gate is CLOSED** — Phase 7 drill D4 (2026-07-24) ran the 1-vs-1 crash-the-oldest survival drill on the three-mesh docker-dev rig and every survivor stayed Up, never self-downing (`c50ebcf7`; `docs/plans/2026-07-24-mesh-phase7-failover-drills.md`).
- **The Primary is the oldest Up `driver` member, not the role leader.** `RedundancyStateActor` previously used `ClusterState.RoleLeader("driver")` (lowest *address*), while `ClusterSingletonManager` places singletons on the *oldest*. They agree only on a freshly-formed cluster and diverge after any restart, so the Primary-gated data plane (writes, alarm acks, alerts emit, alarm-history drain) could enable on a node not hosting the singletons. `NodeRedundancyState.IsRoleLeaderForDriver` was renamed **`IsDriverPrimary`**. - **The Primary is the oldest Up `driver` member, not the role leader.** `RedundancyStateActor` previously used `ClusterState.RoleLeader("driver")` (lowest *address*), while `ClusterSingletonManager` places singletons on the *oldest*. They agree only on a freshly-formed cluster and diverge after any restart, so the Primary-gated data plane (writes, alarm acks, alerts emit, alarm-history drain) could enable on a node not hosting the singletons. `NodeRedundancyState.IsRoleLeaderForDriver` was renamed **`IsDriverPrimary`**.
**A failed OPC UA endpoint steps the Primary down (lmxopcua#762).** `IOpcUaEndpointState` now lives in `Commons.OpcUa` so the Runtime can read it: a `Failed` SDK start feeds `EndpointListening=false` into `ServiceLevelCalculator` (→ 0, beats the +10 Primary bonus), and `OpcUaEndpointStepDownActor` makes a failed **Primary** gracefully `Leave` its mesh (the manual-failover mechanism) once another Up driver member has reported its own endpoint listening on the `opcua-endpoint-state` DPS topic. Do **not** "fix" this by making the election skip the node — the Primary must stay where the singletons are. The peer-report guard is what stops a both-endpoints-failed pair from bouncing the Primary (and restarting nodes) forever. See `docs/Redundancy.md` §"Endpoint start failure: Primary step-down".
**A third bootstrap gap closed 2026-07-22 — downing was never the whole story, and the fix changed twice.** Even under `auto-down`, a node cold-starting while its peer was dead **never came Up**: Akka runs `FirstSeedNodeProcess` — the only bootstrap path that can form a *new* cluster when no peer answers `InitJoin` — exclusively when `seed-nodes[0]` is the node’s own address; every other node runs `JoinSeedNodeProcess` and retries `InitJoin` forever. **`Cluster:SeedNodes` is therefore ORDERED: every node that is one of its own seeds lists ITSELF first, partner second** (docker-dev `central-2` was swapped; site nodes list only `central-1` and are legitimately not seeds). `AkkaClusterOptionsValidator` (`AddValidatedOptions`/`ValidateOnStart` in `AddOtOpcUaCluster`) enforces it at boot — **conditionally** (self must be entry 0 only *if* self is in the list, or every site node would refuse to start), matching on `PublicHostname`+`Port`, never the `0.0.0.0` bind address. The earlier fix — a `Cluster:SelfFormAfter` timer calling `Cluster.Join(SelfAddress)` — is **deleted**: it sat outside Akka’s join handshake, so it could not tell "no seed answered" from "a seed answered and the join is in flight", and it islanded a manually-failed-over node live before a TCP reachability guard patched that one shape. Pinned by `SelfFirstSeedBootstrapTests` (real in-process clusters, incl. a falsifiability control proving peer-first ordering never forms) + `AkkaClusterOptionsValidatorTests`. See `docs/Redundancy.md` §"Bootstrap: self-first seed ordering". **A third bootstrap gap closed 2026-07-22 — downing was never the whole story, and the fix changed twice.** Even under `auto-down`, a node cold-starting while its peer was dead **never came Up**: Akka runs `FirstSeedNodeProcess` — the only bootstrap path that can form a *new* cluster when no peer answers `InitJoin` — exclusively when `seed-nodes[0]` is the node’s own address; every other node runs `JoinSeedNodeProcess` and retries `InitJoin` forever. **`Cluster:SeedNodes` is therefore ORDERED: every node that is one of its own seeds lists ITSELF first, partner second** (docker-dev `central-2` was swapped; site nodes list only `central-1` and are legitimately not seeds). `AkkaClusterOptionsValidator` (`AddValidatedOptions`/`ValidateOnStart` in `AddOtOpcUaCluster`) enforces it at boot — **conditionally** (self must be entry 0 only *if* self is in the list, or every site node would refuse to start), matching on `PublicHostname`+`Port`, never the `0.0.0.0` bind address. The earlier fix — a `Cluster:SelfFormAfter` timer calling `Cluster.Join(SelfAddress)` — is **deleted**: it sat outside Akka’s join handshake, so it could not tell "no seed answered" from "a seed answered and the join is in flight", and it islanded a manually-failed-over node live before a TCP reachability guard patched that one shape. Pinned by `SelfFirstSeedBootstrapTests` (real in-process clusters, incl. a falsifiability control proving peer-first ordering never forms) + `AkkaClusterOptionsValidatorTests`. See `docs/Redundancy.md` §"Bootstrap: self-first seed ordering".
**Simultaneous cold-start splits — the bootstrap guard closes it (opt-in, `Cluster:BootstrapGuard:Enabled`, default OFF).** Self-first on BOTH nodes means when both cold-start at the same instant, each runs `FirstSeedNodeProcess`, times out, and forms its OWN 1-node cluster → two Primaries in one pair (the Phase 6 live gate reproduced this on docker; two separate clusters do NOT auto-merge). The guard prevents it without losing cold-start-alone: the lexicographically **lower** `host:port` node is the preferred founder (self-first, forms immediately); the **higher** node TCP-probes the partner's Akka port up to `PartnerProbeSeconds` (25s) — **reachable ⇒ peer-first (join the founder)**, **unreachable ⇒ self-first (partner dead, form alone)**. The order is chosen BEFORE the single `JoinSeedNodes`, from an explicit reachability signal — it never re-forms mid-handshake (the `SelfFormAfter` failure mode). When on, Akka gets NO config seeds (`BuildClusterOptions` empties `ClusterOptions.SeedNodes`) and `ClusterBootstrapCoordinator` drives the join. Residual trade (accepted, operator-visible via a warning + restart-recovers): the higher node hangs if the founder dies in the probe→join window. docker-dev: **site-a = guard demo** (guard on, serialization removed); **site-b = A/B control** (guard off, `depends_on: service_healthy` serialization). The guard is the production-faithful fix (compose `depends_on` doesn't exist on real VMs). Pinned by `ClusterBootstrapGuardTests` + `ClusterBootstrapCoordinatorTests` (real 2-node clusters, incl. the higher-node-cold-start-alone case). See `docs/Redundancy.md` §"Bootstrap guard". **Simultaneous cold-start splits — the bootstrap guard closes it (opt-in, `Cluster:BootstrapGuard:Enabled`, default OFF).** Self-first on BOTH nodes means when both cold-start at the same instant, each runs `FirstSeedNodeProcess`, times out, and forms its OWN 1-node cluster → two Primaries in one pair (the Phase 6 live gate reproduced this on docker; two separate clusters do NOT auto-merge). The guard prevents it without losing cold-start-alone: the lexicographically **lower** `host:port` node is the preferred founder (self-first, forms immediately); the **higher** node TCP-probes the partner's Akka port up to `PartnerProbeSeconds` (25s) — **reachable ⇒ peer-first (join the founder)**, **unreachable ⇒ self-first (partner dead, form alone)**. The order is chosen BEFORE the single `JoinSeedNodes`, from an explicit reachability signal — it never re-forms mid-handshake (the `SelfFormAfter` failure mode). When on, Akka gets NO config seeds (`BuildClusterOptions` empties `ClusterOptions.SeedNodes`) and `ClusterBootstrapCoordinator` drives the join. Residual trade (accepted, operator-visible via a warning + restart-recovers): the higher node hangs if the founder dies in the probe→join window. docker-dev: **site-a = guard demo** (guard on, serialization removed); **site-b = A/B control** (guard off, `depends_on: service_healthy` serialization). The guard is the production-faithful fix (compose `depends_on` doesn't exist on real VMs). Pinned by `ClusterBootstrapGuardTests` + `ClusterBootstrapCoordinatorTests` (real 2-node clusters, incl. the higher-node-cold-start-alone case). See `docs/Redundancy.md` §"Bootstrap guard".
+37 -1
View File
@@ -30,7 +30,7 @@ The runtime pieces live in:
`RedundancyStateChanged` snapshot, then forwards the result through `RedundancyStateChanged` snapshot, then forwards the result through
`IServiceLevelPublisher` to the SDK's `Server.ServiceLevel` Variable. `IServiceLevelPublisher` to the SDK's `Server.ServiceLevel` Variable.
The four inputs are sourced locally per driver node: The inputs are sourced locally per driver node:
| Input | Source | | Input | Source |
|---|---| |---|---|
@@ -39,12 +39,14 @@ The four inputs are sourced locally per driver node:
| `OpcUaProbeOk` | Result of a peer probing THIS node's OPC UA endpoint: `PeerProbeSupervisor` spawns one `PeerOpcUaProbeActor` per OTHER driver-role peer; each probe publishes `OpcUaProbeResult(probed-node, ok)` on the `redundancy-state` topic; the publish actor consumes only results whose target is itself. Freshness-debounced: absent or stale (>30 s) → `true` (benefit of the doubt — single-node clusters and a departed peer never demote); only an actively-observed RECENT `false` demotes. | | `OpcUaProbeOk` | Result of a peer probing THIS node's OPC UA endpoint: `PeerProbeSupervisor` spawns one `PeerOpcUaProbeActor` per OTHER driver-role peer; each probe publishes `OpcUaProbeResult(probed-node, ok)` on the `redundancy-state` topic; the publish actor consumes only results whose target is itself. Freshness-debounced: absent or stale (>30 s) → `true` (benefit of the doubt — single-node clusters and a departed peer never demote); only an actively-observed RECENT `false` demotes. |
| `Stale` (derived) | `!DbReachable \|\| (now − lastDbHealth.AsOfUtc) > 30 s \|\| (now − snapshotEntry.AsOfUtc) > 30 s`. | | `Stale` (derived) | `!DbReachable \|\| (now − lastDbHealth.AsOfUtc) > 30 s \|\| (now − snapshotEntry.AsOfUtc) > 30 s`. |
| `IsDriverPrimary` | The local node's entry in the `RedundancyStateChanged` snapshot from `RedundancyStateActor`. | | `IsDriverPrimary` | The local node's entry in the `RedundancyStateChanged` snapshot from `RedundancyStateActor`. |
| `EndpointListening` | This node's OWN `IOpcUaEndpointState` (`Commons.OpcUa`), written by `OtOpcUaServerHostedService`: `false` only when the SDK start is recorded `Failed` (`NotStarted`/`Stopped` are boot/shutdown windows, not faults). Read on every recompute. A LOCAL fact on purpose — when another process holds port 4840, the peer TCP probe behind `OpcUaProbeOk` connects to *that* process and reports Ok. lmxopcua#762. |
The resulting truth table (all tiers are now reachable at runtime): The resulting truth table (all tiers are now reachable at runtime):
| Tier | Byte | Condition | | Tier | Byte | Condition |
|---|---|---| |---|---|---|
| Down / Detached | 0 | Member status is not `Up` or `Joining` (leaving, removed, exiting), OR node has no `driver` role (Detached). Published immediately — a starting or detached node never leaves the SDK default 255. | | Down / Detached | 0 | Member status is not `Up` or `Joining` (leaving, removed, exiting), OR node has no `driver` role (Detached). Published immediately — a starting or detached node never leaves the SDK default 255. |
| Endpoint failed | 0 | This node's OPC UA SDK failed to start (`EndpointListening=false`). Wins over every other input, including the +10 Primary bonus; the legacy role-only seam applies it too. See [Endpoint start failure: Primary step-down](#endpoint-start-failure-primary-step-down). |
| Critically degraded | 100 | ConfigDb unreachable AND data is stale. | | Critically degraded | 100 | ConfigDb unreachable AND data is stale. |
| Stale | 200 | Data stale but ConfigDb reachable. | | Stale | 200 | Data stale but ConfigDb reachable. |
| Healthy follower | 240 | DB reachable + OPC UA probe ok + not stale + not the driver Primary. | | Healthy follower | 240 | DB reachable + OPC UA probe ok + not stale + not the driver Primary. |
@@ -384,6 +386,40 @@ Primary — there is no longer a whole-mesh vs. per-cluster ambiguity to caveat.
> testable — deferred to Phase 7's failover drill. See > testable — deferred to Phase 7's failover drill. See
> `docs/plans/2026-07-21-per-cluster-mesh-design.md` §6.2 / Phase 0a. > `docs/plans/2026-07-21-per-cluster-mesh-design.md` §6.2 / Phase 0a.
## Endpoint start failure: Primary step-down
lmxopcua#762 (follow-up F-C1 of review-remediation lane C3). A fused admin+driver node whose OPC UA SDK fails to
start (port 4840 taken, PKI unwritable, a bad `OpcUa:*` override) keeps running so the operator keeps the AdminUI
(a driver-only node instead rethrows and exits). Before #762 it also stayed **Primary** — the Primary-gated data
plane stayed on a node no client could reach while the reachable peer refused writes as Secondary.
Now two things happen:
1. **ServiceLevel 0.** `OpcUaPublishActor` feeds `EndpointListening=false` into `ServiceLevelCalculator`
(above). Nothing can actually read it over OPC UA — the SDK is not running — but the metric and every
in-process consumer see the truth instead of 250/240.
2. **The Primary moves.** `OpcUaEndpointStepDownActor` (`Runtime.Redundancy`, one per driver node, spawned when
the host registers `IOpcUaEndpointState`) makes a failed Primary **leave its mesh** — exactly the
[manual failover](#manual-failover) mechanism, triggered automatically. The election is deliberately *not*
bent to skip the node: the Primary must stay where `ClusterSingletonManager` places the singletons. The node
hands its singletons to the peer, exits, is restarted by its supervisor, and rejoins as the youngest member —
a Secondary at ServiceLevel 0, with its AdminUI back up. The peer advertises 250 and the `/health/active`
tier follows it.
The step-down fires only when **all** hold, re-checked every 5 s:
| Guard | Why |
|---|---|
| Own endpoint is `Failed` | `NotStarted`/`Stopped` are boot/shutdown windows. |
| This node is the driver Primary — `ClusterActiveNode.SelfIsActive(cluster, "driver")` | The same primitive the election delegates to, so the two cannot disagree. A failed Secondary has nothing to hand over. |
| Another Up driver member reported **its own** endpoint listening within the last 20 s | Every driver node publishes an `EndpointReport(node, listening)` on the `opcua-endpoint-state` DistributedPubSub topic each tick. Without this guard a pair whose endpoints **both** failed would bounce the Primary between them forever (each leave restarts a node), and a lone node would shut itself down with nowhere to fail over to. A peer on an older build never reports, so it is never handed the role — the safe direction during a rolling upgrade. |
While a guard holds the role, the node logs one Warning naming the reason (repeated only when the reason changes).
Pinned by `OpcUaEndpointStepDownActorTests` (real two-node mesh: holds while the peer is not listening, then
leaves and the peer becomes the oldest Up driver; a failed Secondary and a healthy Primary never leave),
`OpcUaPublishActorEndpointStateTests` and `ServiceLevelCalculatorTests`. The docker-dev gate (port 4840 taken
⇒ `/health/ready` 503) is lmxopcua#761.
## Bootstrap: self-first seed ordering ## Bootstrap: self-first seed ordering
Listing both peers in `SeedNodes` does **not** mean either node can cold-start alone — and neither does any Listing both peers in `SeedNodes` does **not** mean either node can cold-start alone — and neither does any
@@ -2,12 +2,24 @@ using Akka.Cluster;
namespace ZB.MOM.WW.OtOpcUa.Cluster.Redundancy; namespace ZB.MOM.WW.OtOpcUa.Cluster.Redundancy;
/// <summary>The health facts <see cref="ServiceLevelCalculator.Compute"/> tiers a node on.</summary>
/// <param name="MemberState">This node's Akka cluster member status.</param>
/// <param name="DbReachable">Whether the config DB answered (constant <c>true</c> on a DB-less driver node).</param>
/// <param name="OpcUaProbeOk">A peer's recent TCP probe verdict about this node's OPC UA port (benefit of the
/// doubt when there is none). Observes the PORT only: when another process holds it, the probe succeeds against
/// that process — which is why <paramref name="EndpointListening"/> exists.</param>
/// <param name="Stale">Whether this node is running behind on config.</param>
/// <param name="IsDriverPrimary">Whether this node is the driver Primary (oldest Up driver member).</param>
/// <param name="EndpointListening">This node's OWN view of its OPC UA endpoint: <c>false</c> when the SDK failed
/// to start (<c>IOpcUaEndpointState</c> reports <c>Failed</c>). Defaults to <c>true</c> so a caller with no
/// endpoint fact (a harness, an admin-only node) keeps the previous tiering. lmxopcua#762.</param>
public readonly record struct NodeHealthInputs( public readonly record struct NodeHealthInputs(
MemberStatus MemberState, MemberStatus MemberState,
bool DbReachable, bool DbReachable,
bool OpcUaProbeOk, bool OpcUaProbeOk,
bool Stale, bool Stale,
bool IsDriverPrimary); bool IsDriverPrimary,
bool EndpointListening = true);
/// <summary> /// <summary>
/// Pure ServiceLevel computation per design §6. Output range 0–255, where higher = "more /// Pure ServiceLevel computation per design §6. Output range 0–255, where higher = "more
@@ -16,6 +28,8 @@ public readonly record struct NodeHealthInputs(
/// ///
/// Tiering: /// Tiering:
/// - Member not Up/Joining: 0 (cluster cannot trust this node). /// - Member not Up/Joining: 0 (cluster cannot trust this node).
/// - OPC UA endpoint not listening (the SDK failed to start on this node): 0, with no Primary bonus — the node
/// cannot serve a single client, so it must never out-rank a peer that can (lmxopcua#762).
/// - DB reachable + OPC UA probe ok + not stale: 240 (full service). /// - DB reachable + OPC UA probe ok + not stale: 240 (full service).
/// - Stale config (DB reachable or not, OPC UA probe state ignored): 100 or 200 depending on DB. /// - Stale config (DB reachable or not, OPC UA probe state ignored): 100 or 200 depending on DB.
/// - +10 bonus when this node is the driver Primary (the oldest Up member with the "driver" role — /// - +10 bonus when this node is the driver Primary (the oldest Up member with the "driver" role —
@@ -32,6 +46,12 @@ public static class ServiceLevelCalculator
if (h.MemberState is not (MemberStatus.Up or MemberStatus.Joining)) if (h.MemberState is not (MemberStatus.Up or MemberStatus.Joining))
return 0; return 0;
// lmxopcua#762: a node whose own SDK failed to start serves nothing, whatever its DB, probe, staleness or
// Primary status say. Checked before the tiering so no input can lift it — in particular not the +10
// Primary bonus, which is exactly what used to keep a dead endpoint advertising 250.
if (!h.EndpointListening)
return 0;
var basis = (h.DbReachable, h.OpcUaProbeOk, h.Stale) switch var basis = (h.DbReachable, h.OpcUaProbeOk, h.Stale) switch
{ {
(true, true, false) => 240, (true, true, false) => 240,
@@ -1,6 +1,6 @@
namespace ZB.MOM.WW.OtOpcUa.Host.OpcUa; namespace ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
/// <summary>Lifecycle of this node's OPC UA endpoint as <see cref="OtOpcUaServerHostedService"/> drives it.</summary> /// <summary>Lifecycle of this node's OPC UA endpoint as the Host's <c>OtOpcUaServerHostedService</c> drives it.</summary>
public enum OpcUaEndpointStatus public enum OpcUaEndpointStatus
{ {
/// <summary>The hosted service has not run <c>StartAsync</c> yet.</summary> /// <summary>The hosted service has not run <c>StartAsync</c> yet.</summary>
@@ -23,10 +23,18 @@ public enum OpcUaEndpointStatus
/// connection-refused from the node the fleet called healthy. /// connection-refused from the node the fleet called healthy.
/// </summary> /// </summary>
/// <remarks> /// <remarks>
/// Consumed by <c>OpcUaEndpointHealthCheck</c> (Ready + Active tiers). The redundancy election and the /// <para>
/// ServiceLevel calculator do not read it yet — they live outside the Host (<c>Cluster</c> / /// Lives in Commons (not the Host, where it started) so the Runtime can read it. Three consumers:
/// <c>Runtime</c>); feeding them is the Wave 3 follow-up the lane report names. Until then a fused /// </para>
/// node whose SDK failed is visible (503) but still Primary-eligible. /// <list type="bullet">
/// <item><c>OpcUaEndpointHealthCheck</c> (Host) — Ready tier everywhere, Active tier on a driver-only node.</item>
/// <item><c>OpcUaPublishActor</c> (Runtime) — a <see cref="OpcUaEndpointStatus.Failed"/> endpoint feeds
/// <c>EndpointListening=false</c> into <c>ServiceLevelCalculator</c>, so the node computes ServiceLevel 0
/// instead of 250/240 (lmxopcua#762).</item>
/// <item><c>OpcUaEndpointStepDownActor</c> (Runtime) — a driver Primary whose endpoint failed gracefully
/// leaves its mesh once a peer reports a listening endpoint, so the peer becomes the oldest Up driver
/// member and therefore Primary (lmxopcua#762).</item>
/// </list>
/// </remarks> /// </remarks>
public interface IOpcUaEndpointState public interface IOpcUaEndpointState
{ {
@@ -37,7 +45,8 @@ public interface IOpcUaEndpointState
string? FailureReason { get; } string? FailureReason { get; }
} }
/// <summary>Mutable <see cref="IOpcUaEndpointState"/>; written by the hosted service, read by the health check.</summary> /// <summary>Mutable <see cref="IOpcUaEndpointState"/>; written by the hosted service, read by the health check, the
/// ServiceLevel computation and the Primary step-down.</summary>
public sealed class OpcUaEndpointState : IOpcUaEndpointState public sealed class OpcUaEndpointState : IOpcUaEndpointState
{ {
private volatile OpcUaEndpointStatus _status = OpcUaEndpointStatus.NotStarted; private volatile OpcUaEndpointStatus _status = OpcUaEndpointStatus.NotStarted;
@@ -11,7 +11,7 @@ using ZB.MOM.WW.LocalDb.Replication;
using ZB.MOM.WW.OtOpcUa.Cluster; using ZB.MOM.WW.OtOpcUa.Cluster;
using ZB.MOM.WW.OtOpcUa.Configuration; using ZB.MOM.WW.OtOpcUa.Configuration;
using ZB.MOM.WW.OtOpcUa.Host.Configuration; using ZB.MOM.WW.OtOpcUa.Host.Configuration;
using ZB.MOM.WW.OtOpcUa.Host.OpcUa; using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
namespace ZB.MOM.WW.OtOpcUa.Host.Health; namespace ZB.MOM.WW.OtOpcUa.Host.Health;
@@ -45,11 +45,12 @@ public static class HealthEndpoints
// The ACTIVE tier depends on the roles (coordinator ruling, fix round 1 / I6): // The ACTIVE tier depends on the roles (coordinator ruling, fix round 1 / I6):
// - driver-only: Active too — a site node's whole purpose is its OPC UA endpoint, so nothing may // - driver-only: Active too — a site node's whole purpose is its OPC UA endpoint, so nothing may
// route to it; // route to it;
// - fused admin+driver: NOT Active, until the ServiceLevel calculator consumes IOpcUaEndpointState // - fused admin+driver: NOT Active. Traefik routes the AdminUI on /health/active, so putting the probe
// (follow-up F-C1). Traefik routes the AdminUI on /health/active, so putting the probe there would // there would take the operator's only console out of rotation precisely when they need it to
// take the operator's only console out of rotation precisely when they need it to diagnose the // diagnose the failed endpoint. The OPC UA failover is handled by the Runtime instead (F-C1,
// failed endpoint, while the OPC UA failover (ServiceLevel) still would not move. Once F-C1 lands // lmxopcua#762): the failed node computes ServiceLevel 0 and, when it is the Primary and its peer
// the peer takes Primary and the active tier follows it on its own. // reports a listening endpoint, OpcUaEndpointStepDownActor makes it leave its mesh — the peer becomes
// the oldest Up member, so the active tier follows it on its own.
// Factory-registered so the state singleton is resolved per probe. // Factory-registered so the state singleton is resolved per probe.
if (hasDriver) if (hasDriver)
{ {
@@ -1,5 +1,5 @@
using Microsoft.Extensions.Diagnostics.HealthChecks; using Microsoft.Extensions.Diagnostics.HealthChecks;
using ZB.MOM.WW.OtOpcUa.Host.OpcUa; using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
namespace ZB.MOM.WW.OtOpcUa.Host.Health; namespace ZB.MOM.WW.OtOpcUa.Host.Health;
@@ -12,8 +12,9 @@ namespace ZB.MOM.WW.OtOpcUa.Host.Health;
/// <remarks> /// <remarks>
/// Registered on driver-role nodes only (<see cref="HealthEndpoints.AddOtOpcUaHealth"/>); an /// Registered on driver-role nodes only (<see cref="HealthEndpoints.AddOtOpcUaHealth"/>); an
/// admin-only node serves no OPC UA endpoint and must not be degraded by one it never starts. On a fused /// admin-only node serves no OPC UA endpoint and must not be degraded by one it never starts. On a fused
/// admin+driver node it is Ready-only until ServiceLevel consumes the fact (F-C1), because the Active tier /// admin+driver node it is Ready-only, because the Active tier is what Traefik routes the AdminUI on; the
/// is what Traefik routes the AdminUI on. /// Primary moves through the Runtime instead — ServiceLevel 0 plus <c>OpcUaEndpointStepDownActor</c>
/// (lmxopcua#762).
/// </remarks> /// </remarks>
public sealed class OpcUaEndpointHealthCheck : IHealthCheck public sealed class OpcUaEndpointHealthCheck : IHealthCheck
{ {
@@ -241,6 +241,8 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
/// <c>DbReachable</c> is fed constant <c>true</c> and only snapshot staleness can demote. See /// <c>DbReachable</c> is fed constant <c>true</c> and only snapshot staleness can demote. See
/// <see cref="RecomputeServiceLevel"/>.</summary> /// <see cref="RecomputeServiceLevel"/>.</summary>
private readonly bool _dbLess; private readonly bool _dbLess;
/// <summary>This node's OPC UA endpoint lifecycle fact (lmxopcua#762). Null = no fact (harness) = listening.</summary>
private readonly IOpcUaEndpointState? _endpointState;
private readonly TimeSpan _staleWindow; private readonly TimeSpan _staleWindow;
private readonly TimeSpan _probeFreshnessWindow; private readonly TimeSpan _probeFreshnessWindow;
private readonly TimeSpan _healthTickInterval; private readonly TimeSpan _healthTickInterval;
@@ -306,6 +308,9 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
/// ServiceLevel DB-reachability input is retired (<c>DbReachable</c> treated constant-true, only /// ServiceLevel DB-reachability input is retired (<c>DbReachable</c> treated constant-true, only
/// staleness demotes). No <paramref name="dbHealthProbe"/> is wired on such a node.</param> /// staleness demotes). No <paramref name="dbHealthProbe"/> is wired on such a node.</param>
/// <param name="dataPlaneAuthority">The optional sink that receives the API keys and ACL grants carried by each applied artifact.</param> /// <param name="dataPlaneAuthority">The optional sink that receives the API keys and ACL grants carried by each applied artifact.</param>
/// <param name="endpointState">This node's OPC UA endpoint lifecycle fact (lmxopcua#762). A
/// <see cref="OpcUaEndpointStatus.Failed"/> endpoint computes ServiceLevel 0. Null (harnesses, a node with no
/// hosted SDK) is treated as listening.</param>
/// <returns>The configured <see cref="Props"/> for creating this actor, pinned to the /// <returns>The configured <see cref="Props"/> for creating this actor, pinned to the
/// <see cref="DispatcherId"/> dispatcher.</returns> /// <see cref="DispatcherId"/> dispatcher.</returns>
public static Props Props( public static Props Props(
@@ -320,7 +325,8 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
TimeSpan? healthTickInterval = null, TimeSpan? healthTickInterval = null,
bool dbLess = false, bool dbLess = false,
IDataPlaneAuthoritySink? dataPlaneAuthority = null, IDataPlaneAuthoritySink? dataPlaneAuthority = null,
Func<DateTime>? utcNow = null) => Func<DateTime>? utcNow = null,
IOpcUaEndpointState? endpointState = null) =>
Akka.Actor.Props.Create(() => new OpcUaPublishActor( Akka.Actor.Props.Create(() => new OpcUaPublishActor(
sink ?? NullOpcUaAddressSpaceSink.Instance, sink ?? NullOpcUaAddressSpaceSink.Instance,
serviceLevel ?? NullServiceLevelPublisher.Instance, serviceLevel ?? NullServiceLevelPublisher.Instance,
@@ -334,7 +340,8 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
healthTickInterval, healthTickInterval,
dbLess, dbLess,
dataPlaneAuthority, dataPlaneAuthority,
utcNow)).WithDispatcher(DispatcherId); utcNow,
endpointState)).WithDispatcher(DispatcherId);
/// <summary>Test-only Props that omits the pinned-dispatcher requirement and skips the /// <summary>Test-only Props that omits the pinned-dispatcher requirement and skips the
/// DPS subscribe so unit tests can spin up the actor on a vanilla TestKit cluster.</summary> /// DPS subscribe so unit tests can spin up the actor on a vanilla TestKit cluster.</summary>
@@ -359,6 +366,9 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
/// ServiceLevel DB-reachability input is retired (<c>DbReachable</c> treated constant-true, only /// ServiceLevel DB-reachability input is retired (<c>DbReachable</c> treated constant-true, only
/// staleness demotes). No <paramref name="dbHealthProbe"/> is wired on such a node.</param> /// staleness demotes). No <paramref name="dbHealthProbe"/> is wired on such a node.</param>
/// <param name="dataPlaneAuthority">The optional sink that receives the API keys and ACL grants carried by each applied artifact.</param> /// <param name="dataPlaneAuthority">The optional sink that receives the API keys and ACL grants carried by each applied artifact.</param>
/// <param name="endpointState">This node's OPC UA endpoint lifecycle fact (lmxopcua#762). A
/// <see cref="OpcUaEndpointStatus.Failed"/> endpoint computes ServiceLevel 0. Null (harnesses, a node with no
/// hosted SDK) is treated as listening.</param>
/// <returns>The configured <see cref="Props"/> for creating this actor in tests, without dispatcher /// <returns>The configured <see cref="Props"/> for creating this actor in tests, without dispatcher
/// pinning or the DPS subscribe.</returns> /// pinning or the DPS subscribe.</returns>
public static Props PropsForTests( public static Props PropsForTests(
@@ -374,7 +384,8 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
TimeSpan? healthTickInterval = null, TimeSpan? healthTickInterval = null,
bool dbLess = false, bool dbLess = false,
IDataPlaneAuthoritySink? dataPlaneAuthority = null, IDataPlaneAuthoritySink? dataPlaneAuthority = null,
Func<DateTime>? utcNow = null) => Func<DateTime>? utcNow = null,
IOpcUaEndpointState? endpointState = null) =>
Akka.Actor.Props.Create(() => new OpcUaPublishActor( Akka.Actor.Props.Create(() => new OpcUaPublishActor(
sink ?? NullOpcUaAddressSpaceSink.Instance, sink ?? NullOpcUaAddressSpaceSink.Instance,
serviceLevel ?? NullServiceLevelPublisher.Instance, serviceLevel ?? NullServiceLevelPublisher.Instance,
@@ -388,7 +399,8 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
healthTickInterval, healthTickInterval,
dbLess, dbLess,
dataPlaneAuthority, dataPlaneAuthority,
utcNow)); utcNow,
endpointState));
/// <summary>Test-only Props exposing the deferred-write bound and the plan-step seams.</summary> /// <summary>Test-only Props exposing the deferred-write bound and the plan-step seams.</summary>
internal static Props PropsForTests( internal static Props PropsForTests(
@@ -402,7 +414,7 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
// seam-carrying constructor is internal so the test seams stay off the public API (review M3). // seam-carrying constructor is internal so the test seams stay off the public API (review M3).
Akka.Actor.Props.CreateBy(new TestProducer(() => new OpcUaPublishActor( Akka.Actor.Props.CreateBy(new TestProducer(() => new OpcUaPublishActor(
sink, NullServiceLevelPublisher.Instance, false, localNode, null, applier, null, null, null, null, sink, NullServiceLevelPublisher.Instance, false, localNode, null, applier, null, null, null, null,
false, dataPlaneAuthority, null, deferredWriteCapacity, hooks))); false, dataPlaneAuthority, null, deferredWriteCapacity, hooks, endpointState: null)));
/// <summary>Produces the actor through its internal (seam-carrying) constructor, for <see cref="PropsForTests(IOpcUaAddressSpaceSink, AddressSpaceApplier?, TestHooks?, int, NodeId?, IDataPlaneAuthoritySink?)"/>.</summary> /// <summary>Produces the actor through its internal (seam-carrying) constructor, for <see cref="PropsForTests(IOpcUaAddressSpaceSink, AddressSpaceApplier?, TestHooks?, int, NodeId?, IDataPlaneAuthoritySink?)"/>.</summary>
private sealed class TestProducer(Func<OpcUaPublishActor> create) : IIndirectActorProducer private sealed class TestProducer(Func<OpcUaPublishActor> create) : IIndirectActorProducer
@@ -434,6 +446,7 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
/// ServiceLevel DB-reachability input is retired (<c>DbReachable</c> treated constant-true, only /// ServiceLevel DB-reachability input is retired (<c>DbReachable</c> treated constant-true, only
/// staleness demotes). No <paramref name="dbHealthProbe"/> is wired on such a node.</param> /// staleness demotes). No <paramref name="dbHealthProbe"/> is wired on such a node.</param>
/// <param name="dataPlaneAuthority">The optional sink that receives the API keys and ACL grants carried by each applied artifact.</param> /// <param name="dataPlaneAuthority">The optional sink that receives the API keys and ACL grants carried by each applied artifact.</param>
/// <param name="endpointState">This node's OPC UA endpoint lifecycle fact (lmxopcua#762); null is treated as listening.</param>
public OpcUaPublishActor( public OpcUaPublishActor(
IOpcUaAddressSpaceSink sink, IOpcUaAddressSpaceSink sink,
IServiceLevelPublisher serviceLevel, IServiceLevelPublisher serviceLevel,
@@ -447,15 +460,18 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
TimeSpan? healthTickInterval = null, TimeSpan? healthTickInterval = null,
bool dbLess = false, bool dbLess = false,
IDataPlaneAuthoritySink? dataPlaneAuthority = null, IDataPlaneAuthoritySink? dataPlaneAuthority = null,
Func<DateTime>? utcNow = null) Func<DateTime>? utcNow = null,
IOpcUaEndpointState? endpointState = null)
: this(sink, serviceLevel, subscribeRedundancyTopic, localNode, dbFactory, applier, dbHealthProbe, staleWindow, : this(sink, serviceLevel, subscribeRedundancyTopic, localNode, dbFactory, applier, dbHealthProbe, staleWindow,
probeFreshnessWindow, healthTickInterval, dbLess, dataPlaneAuthority, utcNow, DefaultDeferredWriteCapacity, hooks: null) probeFreshnessWindow, healthTickInterval, dbLess, dataPlaneAuthority, utcNow, DefaultDeferredWriteCapacity, hooks: null,
endpointState)
{ {
} }
/// <summary>The full constructor: also the deferred-write bound and the test seams (internal — tests only).</summary> /// <summary>The full constructor: also the deferred-write bound and the test seams (internal — tests only).</summary>
/// <param name="deferredWriteCapacity">The bound on distinct nodes whose writes are held for replay while a rebuild plan is in flight.</param> /// <param name="deferredWriteCapacity">The bound on distinct nodes whose writes are held for replay while a rebuild plan is in flight.</param>
/// <param name="hooks">Test seams for the off-thread plan step; null in production.</param> /// <param name="hooks">Test seams for the off-thread plan step; null in production.</param>
/// <param name="endpointState">This node's OPC UA endpoint lifecycle fact (lmxopcua#762); null is treated as listening.</param>
internal OpcUaPublishActor( internal OpcUaPublishActor(
IOpcUaAddressSpaceSink sink, IOpcUaAddressSpaceSink sink,
IServiceLevelPublisher serviceLevel, IServiceLevelPublisher serviceLevel,
@@ -471,7 +487,8 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
IDataPlaneAuthoritySink? dataPlaneAuthority, IDataPlaneAuthoritySink? dataPlaneAuthority,
Func<DateTime>? utcNow, Func<DateTime>? utcNow,
int deferredWriteCapacity, int deferredWriteCapacity,
TestHooks? hooks) TestHooks? hooks,
IOpcUaEndpointState? endpointState)
{ {
_deferred = new DeferredWrites(Math.Max(1, deferredWriteCapacity)); _deferred = new DeferredWrites(Math.Max(1, deferredWriteCapacity));
_hooks = hooks; _hooks = hooks;
@@ -486,6 +503,7 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
_applier = applier; _applier = applier;
_dbHealthProbe = dbHealthProbe; _dbHealthProbe = dbHealthProbe;
_dbLess = dbLess; _dbLess = dbLess;
_endpointState = endpointState;
_staleWindow = staleWindow ?? TimeSpan.FromSeconds(30); _staleWindow = staleWindow ?? TimeSpan.FromSeconds(30);
_probeFreshnessWindow = probeFreshnessWindow ?? TimeSpan.FromSeconds(30); _probeFreshnessWindow = probeFreshnessWindow ?? TimeSpan.FromSeconds(30);
_healthTickInterval = healthTickInterval ?? TimeSpan.FromSeconds(5); _healthTickInterval = healthTickInterval ?? TimeSpan.FromSeconds(5);
@@ -1536,6 +1554,15 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
return verdict.Ok; return verdict.Ok;
} }
/// <summary>The OPC UA endpoint input for the calculator (lmxopcua#762): <c>false</c> only when this node's own
/// SDK start is recorded as <see cref="OpcUaEndpointStatus.Failed"/>. <see cref="OpcUaEndpointStatus.NotStarted"/>
/// and <see cref="OpcUaEndpointStatus.Stopped"/> are boot and shutdown windows, not faults, and a node with no
/// endpoint fact at all (a harness) is treated as listening. Read on every recompute — the periodic health tick
/// and every redundancy snapshot (heartbeated every 10 s) — so a change is picked up without a notification.
/// It is a LOCAL fact on purpose: the peer TCP probe behind <see cref="OpcUaProbeOk"/> cannot see a failed start
/// when another process holds the port, because its connect then succeeds against that process.</summary>
private bool EndpointListening() => _endpointState?.Status != OpcUaEndpointStatus.Failed;
/// <summary> /// <summary>
/// Computes the local OPC UA ServiceLevel and routes it through <see cref="ServiceLevelChanged"/> /// Computes the local OPC UA ServiceLevel and routes it through <see cref="ServiceLevelChanged"/>
/// (the dedup/publish/metric handler). The full <see cref="ServiceLevelCalculator"/> path is /// (the dedup/publish/metric handler). The full <see cref="ServiceLevelCalculator"/> path is
@@ -1575,7 +1602,8 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
DbReachable: true, DbReachable: true,
OpcUaProbeOk: OpcUaProbeOk(), OpcUaProbeOk: OpcUaProbeOk(),
Stale: (utcNow - entry.AsOfUtc) > _staleWindow, Stale: (utcNow - entry.AsOfUtc) > _staleWindow,
IsDriverPrimary: entry.IsDriverPrimary); IsDriverPrimary: entry.IsDriverPrimary,
EndpointListening: EndpointListening());
Self.Tell(new ServiceLevelChanged(ServiceLevelCalculator.Compute(dbLessInputs))); Self.Tell(new ServiceLevelChanged(ServiceLevelCalculator.Compute(dbLessInputs)));
return; return;
} }
@@ -1585,7 +1613,9 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
// is the bootstrap value until the first DbHealthStatus lands. // is the bootstrap value until the first DbHealthStatus lands.
if (_dbHealthProbe is null || _lastDbHealth is null) if (_dbHealthProbe is null || _lastDbHealth is null)
{ {
Self.Tell(new ServiceLevelChanged(LegacyRoleOnly(entry))); // The role-only seam has no health inputs, but a failed endpoint still wins (lmxopcua#762): the
// legacy switch would otherwise hand a node that serves nothing 240 as Primary.
Self.Tell(new ServiceLevelChanged(EndpointListening() ? LegacyRoleOnly(entry) : (byte)0));
return; return;
} }
@@ -1597,7 +1627,8 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
Stale: !_lastDbHealth.Reachable Stale: !_lastDbHealth.Reachable
|| (now - _lastDbHealth.AsOfUtc) > _staleWindow || (now - _lastDbHealth.AsOfUtc) > _staleWindow
|| (now - entry.AsOfUtc) > _staleWindow, || (now - entry.AsOfUtc) > _staleWindow,
IsDriverPrimary: entry.IsDriverPrimary); IsDriverPrimary: entry.IsDriverPrimary,
EndpointListening: EndpointListening());
Self.Tell(new ServiceLevelChanged(ServiceLevelCalculator.Compute(inputs))); Self.Tell(new ServiceLevelChanged(ServiceLevelCalculator.Compute(inputs)));
} }
@@ -0,0 +1,226 @@
using Akka.Actor;
using Akka.Cluster;
using Akka.Cluster.Tools.PublishSubscribe;
using Akka.Event;
using ZB.MOM.WW.Health.Akka;
using ZB.MOM.WW.OtOpcUa.Cluster;
using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
using ZB.MOM.WW.OtOpcUa.Commons.Types;
namespace ZB.MOM.WW.OtOpcUa.Runtime.Redundancy;
/// <summary>
/// Steps this node down from driver Primary while its own OPC UA endpoint is not listening (lmxopcua#762).
/// </summary>
/// <remarks>
/// <para>
/// <b>Why.</b> A fused admin+driver node whose SDK failed to start (port 4840 taken, PKI unwritable, a bad
/// <c>OpcUa:*</c> override) keeps running so the operator still has the AdminUI. Before this actor it also
/// stayed Primary: the oldest Up driver member, so the Primary-gated data plane (writes, alarm acks, the
/// alerts emit, the alarm-history drain, HistoryUpdate) stayed enabled on a node no client can reach, while
/// the peer that clients CAN reach refused writes as Secondary.
/// </para>
/// <para>
/// <b>How.</b> The Primary is the oldest Up <c>driver</c> member and must stay exactly where
/// <c>ClusterSingletonManager</c> puts the singletons (see <c>RedundancyStateActor.SelectDriverPrimary</c>),
/// so the election is NOT bent to skip this node — that would split the Primary-gated data plane from the
/// singletons. Instead the node gracefully <b>leaves</b> its mesh, the same mechanism as an operator's manual
/// failover: it hands its singletons to the peer, <c>CoordinatedShutdown</c> runs,
/// <c>ActorSystemTerminationWatchdog</c> exits the process, the service supervisor restarts it, and it rejoins
/// as the youngest member — a Secondary whose ServiceLevel is 0 (the endpoint fact also feeds
/// <c>ServiceLevelCalculator</c>). The peer is now the oldest Up driver member, so it is Primary, hosts the
/// singletons, advertises 250, and the <c>/health/active</c> tier follows it.
/// </para>
/// <para>
/// <b>Guards</b> — every one must hold, checked on each tick:
/// </para>
/// <list type="number">
/// <item>this node's endpoint is <see cref="OpcUaEndpointStatus.Failed"/> (not merely
/// <see cref="OpcUaEndpointStatus.NotStarted"/> or <see cref="OpcUaEndpointStatus.Stopped"/>, which are boot
/// and shutdown windows);</item>
/// <item>this node IS the driver Primary, judged by <see cref="ClusterActiveNode.SelfIsActive"/> — the same
/// primitive the election delegates to, so the two cannot disagree;</item>
/// <item>another Up driver member has RECENTLY reported, on <see cref="Topic"/>, that its own endpoint is
/// listening. Without this a pair whose endpoints BOTH failed would bounce the Primary between them forever,
/// restarting each node in turn — and a single node (no peer) would shut itself down for nothing. A peer on
/// an older build never reports, so it never receives the role: that is the safe direction.</item>
/// </list>
/// <para>
/// Every node also publishes its own <see cref="EndpointReport"/> on each tick, which is how its peer learns
/// whether it can take over. The topic is separate from <c>redundancy-state</c> so the half-dozen actors that
/// subscribe there are not handed a message type they do not handle.
/// </para>
/// </remarks>
public sealed class OpcUaEndpointStepDownActor : ReceiveActor, IWithTimers
{
/// <summary>The DistributedPubSub topic every driver node publishes its <see cref="EndpointReport"/> on.</summary>
public const string Topic = "opcua-endpoint-state";
/// <summary>Default period between self-reports and step-down evaluations.</summary>
public static readonly TimeSpan DefaultCheckInterval = TimeSpan.FromSeconds(5);
/// <summary>Default age beyond which a peer's report is ignored (it went away or stopped reporting).</summary>
public static readonly TimeSpan DefaultReportFreshness = TimeSpan.FromSeconds(20);
/// <summary>One node's statement about its own OPC UA endpoint.</summary>
/// <param name="Node">The reporting node, as <c>host:port</c> of its Akka address.</param>
/// <param name="Listening">Whether the node's SDK is started and serving.</param>
public sealed record EndpointReport(NodeId Node, bool Listening);
/// <summary>Periodic self-tick.</summary>
public sealed class Tick
{
/// <summary>The singleton tick instance.</summary>
public static readonly Tick Instance = new();
private Tick() { }
}
private readonly IOpcUaEndpointState _state;
private readonly TimeSpan _interval;
private readonly TimeSpan _freshness;
private readonly Func<DateTime> _utcNow;
private readonly Akka.Cluster.Cluster _cluster = Akka.Cluster.Cluster.Get(Context.System);
private readonly ILoggingAdapter _log = Context.GetLogger();
private readonly Dictionary<NodeId, (bool Listening, DateTime At)> _peerReports = new();
private readonly NodeId _self;
private bool _leaving;
private string? _lastHoldReason;
/// <summary>Gets or sets the timer scheduler.</summary>
public ITimerScheduler Timers { get; set; } = null!;
/// <summary>Creates the actor's props.</summary>
/// <param name="state">This node's OPC UA endpoint lifecycle fact.</param>
/// <param name="checkInterval">Period between self-reports and evaluations; defaults to <see cref="DefaultCheckInterval"/>.</param>
/// <param name="reportFreshness">Peer-report age limit; defaults to <see cref="DefaultReportFreshness"/>.</param>
/// <param name="utcNow">Clock; defaults to <see cref="DateTime.UtcNow"/>.</param>
/// <returns>The props.</returns>
public static Props Props(
IOpcUaEndpointState state,
TimeSpan? checkInterval = null,
TimeSpan? reportFreshness = null,
Func<DateTime>? utcNow = null) =>
Akka.Actor.Props.Create(() => new OpcUaEndpointStepDownActor(state, checkInterval, reportFreshness, utcNow));
/// <summary>Initializes a new instance of the <see cref="OpcUaEndpointStepDownActor"/> class.</summary>
/// <param name="state">This node's OPC UA endpoint lifecycle fact.</param>
/// <param name="checkInterval">Period between self-reports and evaluations.</param>
/// <param name="reportFreshness">Peer-report age limit.</param>
/// <param name="utcNow">Clock.</param>
public OpcUaEndpointStepDownActor(
IOpcUaEndpointState state,
TimeSpan? checkInterval,
TimeSpan? reportFreshness,
Func<DateTime>? utcNow)
{
_state = state ?? throw new ArgumentNullException(nameof(state));
_interval = checkInterval ?? DefaultCheckInterval;
_freshness = reportFreshness ?? DefaultReportFreshness;
_utcNow = utcNow ?? (() => DateTime.UtcNow);
_self = ToNodeId(_cluster.SelfAddress);
Receive<Tick>(_ => OnTick());
Receive<EndpointReport>(OnReport);
Receive<SubscribeAck>(_ => { });
}
/// <summary>
/// The canonical <c>host:port</c> id of an Akka address — the format <c>RedundancyStateActor.ToNodeId</c> and
/// <c>ClusterRoleInfo</c> use. Derived from the member address on both sides (reporter and reader), so the two
/// always agree.
/// </summary>
/// <param name="address">The member address.</param>
/// <returns>The node id.</returns>
public static NodeId ToNodeId(Address address) =>
NodeId.Parse($"{address.Host ?? string.Empty}:{address.Port ?? 0}");
/// <inheritdoc />
protected override void PreStart()
{
DistributedPubSub.Get(Context.System).Mediator.Tell(new Subscribe(Topic, Self));
Self.Tell(Tick.Instance);
Timers.StartPeriodicTimer("tick", Tick.Instance, _interval);
}
private void OnTick()
{
var listening = _state.Status == OpcUaEndpointStatus.Started;
try
{
DistributedPubSub.Get(Context.System).Mediator.Tell(new Publish(Topic, new EndpointReport(_self, listening)));
}
catch (Exception ex)
{
_log.Debug(ex, "OpcUaEndpointStepDown: could not publish this node's endpoint report");
}
Evaluate();
}
private void OnReport(EndpointReport report)
{
if (report.Node == _self) return; // our own publish echoed back
_peerReports[report.Node] = (report.Listening, _utcNow());
Evaluate();
}
private void Evaluate()
{
if (_leaving) return;
if (_state.Status != OpcUaEndpointStatus.Failed)
{
_lastHoldReason = null;
return;
}
bool isPrimary;
try
{
isPrimary = ClusterActiveNode.SelfIsActive(_cluster, RoleParser.Driver);
}
catch (Exception ex)
{
_log.Debug(ex, "OpcUaEndpointStepDown: cluster state unavailable; not evaluating");
return;
}
if (!isPrimary)
{
HoldOnce("not the driver Primary — nothing to step down from");
return;
}
var now = _utcNow();
var self = _cluster.SelfMember.UniqueAddress;
// Ordinal Member.HasRole on purpose — the same match ClusterActiveNode (and so the election) applies, so a
// candidate here is a node the election would actually name Primary once this one leaves.
var takeover = _cluster.State.Members
.Where(m => m.Status == MemberStatus.Up
&& m.HasRole(RoleParser.Driver)
&& !m.UniqueAddress.Equals(self))
.Select(m => (NodeId?)ToNodeId(m.Address))
.FirstOrDefault(id => _peerReports.TryGetValue(id!.Value, out var r) && r.Listening && now - r.At <= _freshness);
if (takeover is not { } peer)
{
HoldOnce("no Up driver peer reports a listening OPC UA endpoint — staying Primary rather than handing the role to a node that cannot serve either");
return;
}
_leaving = true;
_log.Warning(
"OpcUaEndpointStepDown: this node is the driver Primary but its OPC UA endpoint failed to start ({Reason}); "
+ "peer {Peer} reports a listening endpoint. Leaving the cluster so {Peer} becomes Primary — this node hands "
+ "over its singletons, exits, and is restarted by its supervisor as a Secondary (lmxopcua#762).",
_state.FailureReason ?? "no reason recorded", peer, peer);
_cluster.Leave(_cluster.SelfAddress);
}
private void HoldOnce(string reason)
{
if (reason == _lastHoldReason) return;
_lastHoldReason = reason;
_log.Warning("OpcUaEndpointStepDown: OPC UA endpoint failed to start ({Failure}); {Reason}",
_state.FailureReason ?? "no reason recorded", reason);
}
}
@@ -44,6 +44,7 @@ public static class ServiceCollectionExtensions
public const string DependencyMuxActorName = "dependency-mux"; public const string DependencyMuxActorName = "dependency-mux";
public const string OpcUaPublishActorName = "opcua-publish"; public const string OpcUaPublishActorName = "opcua-publish";
public const string PeerProbeSupervisorName = "peer-probe-supervisor"; public const string PeerProbeSupervisorName = "peer-probe-supervisor";
public const string OpcUaEndpointStepDownActorName = "opcua-endpoint-step-down";
public const string ContinuousHistorizationRecorderActorName = "continuous-historization-recorder"; public const string ContinuousHistorizationRecorderActorName = "continuous-historization-recorder";
/// <summary> /// <summary>
@@ -555,6 +556,10 @@ public static class ServiceCollectionExtensions
// a node with no driver role never registers it and test harnesses build this graph without it; // a node with no driver role never registers it and test harnesses build this graph without it;
// the actor treats null as "nothing consumes authorization inputs" and skips the decode. // the actor treats null as "nothing consumes authorization inputs" and skips the decode.
var dataPlaneAuthority = resolver.GetService<IDataPlaneAuthoritySink>(); var dataPlaneAuthority = resolver.GetService<IDataPlaneAuthoritySink>();
// lmxopcua#762: the "did the OPC UA SDK start?" fact the Host's OtOpcUaServerHostedService maintains.
// Optional: a harness that builds this graph without the hosted service registers none, and both
// consumers then behave as if the endpoint were listening (the pre-#762 behaviour).
var endpointState = resolver.GetService<IOpcUaEndpointState>();
var publishActor = system.ActorOf( var publishActor = system.ActorOf(
OpcUaPublishActor.Props( OpcUaPublishActor.Props(
sink: addressSpaceSink, sink: addressSpaceSink,
@@ -564,10 +569,19 @@ public static class ServiceCollectionExtensions
applier: applier, applier: applier,
dbHealthProbe: dbHealth, dbHealthProbe: dbHealth,
dbLess: dbFactory is null, dbLess: dbFactory is null,
dataPlaneAuthority: dataPlaneAuthority), dataPlaneAuthority: dataPlaneAuthority,
endpointState: endpointState),
OpcUaPublishActorName); OpcUaPublishActorName);
registry.Register<OpcUaPublishActorKey>(publishActor); registry.Register<OpcUaPublishActorKey>(publishActor);
// lmxopcua#762: a driver Primary whose OPC UA endpoint failed to start leaves its mesh once a peer
// reports a listening endpoint, so the peer becomes Primary. Every driver node runs one — it is also
// how this node tells its peer whether IT can take over.
if (endpointState is not null)
{
system.ActorOf(OpcUaEndpointStepDownActor.Props(endpointState), OpcUaEndpointStepDownActorName);
}
// Per-node peer-probe supervisor — keeps one OPC UA TCP probe per OTHER non-Detached // Per-node peer-probe supervisor — keeps one OPC UA TCP probe per OTHER non-Detached
// driver node, so every node is continuously probed by all its peers. OpcUaPublishActor // driver node, so every node is continuously probed by all its peers. OpcUaPublishActor
// consumes the resulting probe verdicts to learn this node's own reachability. // consumes the resulting probe verdicts to learn this node's own reachability.
@@ -23,6 +23,10 @@
<!-- Core LocalDb only (ILocalDb + the connection/transaction seam) — the deployment-artifact <!-- Core LocalDb only (ILocalDb + the connection/transaction seam) — the deployment-artifact
cache lives here, but the sync engine and its gRPC endpoint are the Host's concern. --> cache lives here, but the sync engine and its gRPC endpoint are the Host's concern. -->
<PackageReference Include="ZB.MOM.WW.LocalDb" /> <PackageReference Include="ZB.MOM.WW.LocalDb" />
<!-- ClusterActiveNode — the same "oldest Up member of role X" primitive RedundancyStateActor's election
delegates to. OpcUaEndpointStepDownActor (lmxopcua#762) asks "am I the driver Primary?" through it so
the step-down and the election cannot disagree about which node is Primary. -->
<PackageReference Include="ZB.MOM.WW.Health.Akka" />
</ItemGroup> </ItemGroup>
<ItemGroup> <ItemGroup>
@@ -88,4 +88,32 @@ public sealed class ServiceLevelCalculatorTests
DbReachable: true, OpcUaProbeOk: true, Stale: false, IsDriverPrimary: true)); DbReachable: true, OpcUaProbeOk: true, Stale: false, IsDriverPrimary: true));
((int)sl).ShouldBeLessThanOrEqualTo(255); ((int)sl).ShouldBeLessThanOrEqualTo(255);
} }
/// <summary>
/// lmxopcua#762: a node whose own OPC UA endpoint failed to start computes 0 — whatever every other input
/// says, and in particular without the +10 Primary bonus that used to keep a dead endpoint at 250.
/// </summary>
[Theory]
[InlineData(true, true, false, true)] // would be 250
[InlineData(true, true, false, false)] // would be 240
[InlineData(true, true, true, true)] // would be 210
[InlineData(false, true, true, false)] // would be 100
public void Endpoint_not_listening_returns_zero(bool dbReachable, bool probeOk, bool stale, bool primary)
{
var healthy = ServiceLevelCalculator.Compute(new(MemberStatus.Up,
DbReachable: dbReachable, OpcUaProbeOk: probeOk, Stale: stale, IsDriverPrimary: primary));
healthy.ShouldBeGreaterThan((byte)0, "positive control: the same inputs with a listening endpoint must not already be 0");
var sl = ServiceLevelCalculator.Compute(new(MemberStatus.Up,
DbReachable: dbReachable, OpcUaProbeOk: probeOk, Stale: stale, IsDriverPrimary: primary,
EndpointListening: false));
sl.ShouldBe((byte)0);
}
/// <summary>The endpoint input defaults to listening, so callers with no endpoint fact keep the old tiering.</summary>
[Fact]
public void Endpoint_listening_defaults_to_true()
{
new NodeHealthInputs(MemberStatus.Up, true, true, false, true).EndpointListening.ShouldBeTrue();
}
} }
@@ -0,0 +1,119 @@
using System.Collections.Concurrent;
using Akka.Actor;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Redundancy;
using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
using ZB.MOM.WW.OtOpcUa.Commons.Types;
using ZB.MOM.WW.OtOpcUa.Runtime.Health;
using ZB.MOM.WW.OtOpcUa.Runtime.OpcUa;
using ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness;
namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.OpcUa;
/// <summary>
/// lmxopcua#762: a node whose own OPC UA SDK failed to start must not advertise 250/240. Each test pairs the
/// failed endpoint with a positive control — the SAME actor wiring and snapshot with a started endpoint — so a
/// 0 cannot come from some other demotion (Detached, a missing entry, a peer probe).
/// </summary>
public sealed class OpcUaPublishActorEndpointStateTests : RuntimeActorTestBase
{
/// <summary>DB-less driver node (the survive-alone posture): Primary would be 250, a failed endpoint is 0.</summary>
[Theory]
[InlineData(true, 250)]
[InlineData(false, 240)]
public void DbLess_failed_endpoint_publishes_zero_instead_of_healthy_level(bool primary, byte healthy)
{
PublishedLevelFor(OpcUaEndpointStatus.Started, primary, dbLess: true).ShouldBe(healthy, "positive control");
PublishedLevelFor(OpcUaEndpointStatus.Failed, primary, dbLess: true).ShouldBe((byte)0);
}
/// <summary>DB-backed node on the full calculator path (DB reachable, fresh): same rule.</summary>
[Fact]
public void DbBacked_failed_endpoint_publishes_zero_instead_of_250()
{
PublishedLevelFor(OpcUaEndpointStatus.Started, primary: true, dbLess: false, dbBacked: true).ShouldBe((byte)250, "positive control");
PublishedLevelFor(OpcUaEndpointStatus.Failed, primary: true, dbLess: false, dbBacked: true).ShouldBe((byte)0);
}
/// <summary>The legacy role-only seam (no DB probe wired yet) must not hand a failed endpoint 240 either.</summary>
[Fact]
public void Legacy_role_only_seam_failed_endpoint_publishes_zero_instead_of_240()
{
PublishedLevelFor(OpcUaEndpointStatus.Started, primary: true, dbLess: false).ShouldBe((byte)240, "positive control");
PublishedLevelFor(OpcUaEndpointStatus.Failed, primary: true, dbLess: false).ShouldBe((byte)0);
}
/// <summary>
/// Boot (<see cref="OpcUaEndpointStatus.NotStarted"/>) and shutdown (<see cref="OpcUaEndpointStatus.Stopped"/>)
/// are not faults: only a recorded start FAILURE demotes.
/// </summary>
[Theory]
[InlineData(OpcUaEndpointStatus.NotStarted)]
[InlineData(OpcUaEndpointStatus.Stopped)]
public void Only_a_failed_start_demotes(OpcUaEndpointStatus status)
{
PublishedLevelFor(status, primary: true, dbLess: true).ShouldBe((byte)250);
}
/// <summary>
/// The fact is read on every recompute, so an endpoint that fails after the first healthy publish demotes on
/// the next health tick without any message being sent to the actor.
/// </summary>
[Fact]
public void Endpoint_failing_later_demotes_on_the_next_tick()
{
var publisher = new RecordingPublisher();
var state = new OpcUaEndpointState();
state.MarkStarted();
var local = NodeId.Parse("endpoint-later-node");
var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests(
serviceLevel: publisher, localNode: local, dbLess: true,
healthTickInterval: TimeSpan.FromMilliseconds(25), endpointState: state));
actor.Tell(Snapshot(local, primary: true));
AwaitAssert(() => publisher.Levels.ShouldBe(new byte[] { 250 }), PresenceBudget);
state.MarkFailed("port 4840 in use");
AwaitAssert(() => publisher.Levels.ShouldBe(new byte[] { 250, 0 }), PresenceBudget);
}
private byte PublishedLevelFor(OpcUaEndpointStatus status, bool primary, bool dbLess, bool dbBacked = false)
{
var publisher = new RecordingPublisher();
var state = new OpcUaEndpointState();
switch (status)
{
case OpcUaEndpointStatus.Started: state.MarkStarted(); break;
case OpcUaEndpointStatus.Failed: state.MarkFailed("port 4840 in use"); break;
case OpcUaEndpointStatus.Stopped: state.MarkStopped(); break;
}
var local = NodeId.Parse($"endpoint-{Guid.NewGuid():N}");
// A DB-backed actor needs a probe ref to leave the legacy seam; its Ask is never answered here (the 5-min
// tick keeps it to the one start-up Ask), and a healthy sample is Told directly BEFORE the snapshot so the
// snapshot's recompute — the first publish — runs the full calculator path.
var db = dbBacked ? CreateTestProbe() : null;
var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests(
serviceLevel: publisher, localNode: local, dbLess: dbLess, dbHealthProbe: db?.Ref,
healthTickInterval: TimeSpan.FromMinutes(5), endpointState: state));
if (db is not null)
actor.Tell(new DbHealthProbeActor.DbHealthStatus(true, DateTime.UtcNow, null));
actor.Tell(Snapshot(local, primary));
AwaitAssert(() => publisher.Levels.Length.ShouldBeGreaterThanOrEqualTo(1), PresenceBudget);
Sys.Stop(actor);
return publisher.Levels[0];
}
private static RedundancyStateChanged Snapshot(NodeId local, bool primary) =>
new(new[] { new NodeRedundancyState(local,
primary ? RedundancyRole.Primary : RedundancyRole.Secondary,
IsClusterLeader: primary, IsDriverPrimary: primary, DateTime.UtcNow) }, CorrelationId.NewId());
private sealed class RecordingPublisher : IServiceLevelPublisher
{
private readonly ConcurrentQueue<byte> _levels = new();
public byte[] Levels => _levels.ToArray();
public void Publish(byte serviceLevel) => _levels.Enqueue(serviceLevel);
}
}
@@ -0,0 +1,147 @@
using Akka.Actor;
using Akka.Cluster;
using Akka.TestKit.Xunit2;
using Shouldly;
using Xunit;
using ZB.MOM.WW.Health.Akka;
using ZB.MOM.WW.OtOpcUa.Cluster;
using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
using ZB.MOM.WW.OtOpcUa.Runtime.Redundancy;
using ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness;
namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Redundancy;
/// <summary>
/// lmxopcua#762 on a real two-node driver mesh: a Primary whose OPC UA endpoint failed to start hands the role to
/// its peer by leaving the mesh — but only once the peer reports a listening endpoint of its own.
/// </summary>
/// <remarks>
/// Node <c>A</c> is a second <see cref="TestKit"/> that forms the cluster first, so it is the OLDEST Up driver
/// member — the node the election (<see cref="ClusterActiveNode.OldestUpMember(IEnumerable{Member}, string?)"/>)
/// names Primary. This class's own <see cref="TestKit.Sys"/> is node <c>B</c>, which joins second; it survives
/// every test, so the base class disposes it normally.
/// </remarks>
public sealed class OpcUaEndpointStepDownActorTests : TestKit
{
private const string Hocon = @"
akka {
loglevel = ""WARNING""
test.single-expect-default = 15s
extensions = [
""Akka.Cluster.Tools.PublishSubscribe.DistributedPubSubExtensionProvider, Akka.Cluster.Tools""
]
actor.provider = ""Akka.Cluster.ClusterActorRefProvider, Akka.Cluster""
remote.dot-netty.tcp {
hostname = ""127.0.0.1""
port = 0
}
cluster {
seed-nodes = []
roles = [""driver""]
min-nr-of-members = 1
run-coordinated-shutdown-when-down = off
pub-sub.gossip-interval = 200ms
}
}";
private static readonly TimeSpan Tick = TimeSpan.FromMilliseconds(100);
/// <summary>How long a "does NOT leave" assertion watches — many ticks, and well past pub-sub convergence.</summary>
private static readonly TimeSpan AbsenceWindow = TimeSpan.FromSeconds(3);
private readonly TestKit _nodeA;
public OpcUaEndpointStepDownActorTests() : base(Hocon + TestTimeFactor.Hocon)
{
_nodeA = new TestKit(Hocon + TestTimeFactor.Hocon);
var clusterA = Akka.Cluster.Cluster.Get(_nodeA.Sys);
clusterA.Join(clusterA.SelfAddress);
AwaitCondition(() => UpCount(clusterA) == 1, RuntimeActorTestBase.PresenceBudget);
var clusterB = Akka.Cluster.Cluster.Get(Sys);
clusterB.Join(clusterA.SelfAddress);
AwaitCondition(() => UpCount(clusterA) == 2 && UpCount(clusterB) == 2, RuntimeActorTestBase.PresenceBudget);
// Fixture sanity: A must be the Primary going in, or every assertion below is about the wrong node.
ClusterActiveNode.SelfIsActive(clusterA, RoleParser.Driver).ShouldBeTrue("node A formed the cluster, so it is the oldest Up driver member");
ClusterActiveNode.SelfIsActive(clusterB, RoleParser.Driver).ShouldBeFalse();
}
/// <inheritdoc />
protected override void Dispose(bool disposing)
{
if (disposing) _nodeA.Dispose();
base.Dispose(disposing);
}
/// <summary>
/// The issue's test: A's endpoint failed to start, so A steps down and B becomes Primary. Also pins the
/// ping-pong guard first — while B's own endpoint is not listening, A holds the role (handing it over would
/// only move the outage and restart A for nothing); the moment B reports listening, A leaves.
/// </summary>
[Fact]
public async Task Failed_primary_hands_the_role_to_a_peer_once_the_peer_reports_listening()
{
var stateA = Failed();
var stateB = new OpcUaEndpointState(); // NotStarted: not listening yet
_nodeA.Sys.ActorOf(OpcUaEndpointStepDownActor.Props(stateA, Tick), "step-down");
Sys.ActorOf(OpcUaEndpointStepDownActor.Props(stateB, Tick), "step-down");
// Guard: no listening peer -> A stays Up and Primary.
await Task.Delay(AbsenceWindow);
_nodeA.Sys.WhenTerminated.IsCompleted.ShouldBeFalse("A must not leave while no peer can serve either");
ClusterActiveNode.SelfIsActive(Akka.Cluster.Cluster.Get(_nodeA.Sys), RoleParser.Driver).ShouldBeTrue();
// B's endpoint comes up -> A leaves (its ActorSystem terminates, which in production exits the process)...
stateB.MarkStarted();
await _nodeA.Sys.WhenTerminated.WaitAsync(RuntimeActorTestBase.PresenceBudget);
// ... and B is now the oldest Up driver member, i.e. what RedundancyStateActor elects Primary.
var clusterB = Akka.Cluster.Cluster.Get(Sys);
AwaitCondition(() => ClusterActiveNode.SelfIsActive(clusterB, RoleParser.Driver)
&& clusterB.State.Members.Count(m => m.Status == MemberStatus.Up) == 1,
RuntimeActorTestBase.PresenceBudget);
}
/// <summary>A failed endpoint on the SECONDARY changes nothing: there is no role to hand over.</summary>
[Fact]
public async Task Failed_secondary_does_not_leave()
{
var stateA = new OpcUaEndpointState();
stateA.MarkStarted();
_nodeA.Sys.ActorOf(OpcUaEndpointStepDownActor.Props(stateA, Tick), "step-down");
Sys.ActorOf(OpcUaEndpointStepDownActor.Props(Failed(), Tick), "step-down");
await Task.Delay(AbsenceWindow);
Sys.WhenTerminated.IsCompleted.ShouldBeFalse();
_nodeA.Sys.WhenTerminated.IsCompleted.ShouldBeFalse();
UpCount(Akka.Cluster.Cluster.Get(Sys)).ShouldBe(2);
ClusterActiveNode.SelfIsActive(Akka.Cluster.Cluster.Get(_nodeA.Sys), RoleParser.Driver).ShouldBeTrue();
}
/// <summary>A healthy Primary never leaves, even with a listening peer (positive control for the trigger).</summary>
[Fact]
public async Task Healthy_primary_does_not_leave()
{
var stateA = new OpcUaEndpointState();
stateA.MarkStarted();
var stateB = new OpcUaEndpointState();
stateB.MarkStarted();
_nodeA.Sys.ActorOf(OpcUaEndpointStepDownActor.Props(stateA, Tick), "step-down");
Sys.ActorOf(OpcUaEndpointStepDownActor.Props(stateB, Tick), "step-down");
await Task.Delay(AbsenceWindow);
_nodeA.Sys.WhenTerminated.IsCompleted.ShouldBeFalse();
UpCount(Akka.Cluster.Cluster.Get(Sys)).ShouldBe(2);
}
private static OpcUaEndpointState Failed()
{
var state = new OpcUaEndpointState();
state.MarkFailed("port 4840 in use");
return state;
}
private static int UpCount(Akka.Cluster.Cluster cluster) =>
cluster.State.Members.Count(m => m.Status == MemberStatus.Up);
}