diff --git a/.claude/rules/cluster-topology.md b/.claude/rules/cluster-topology.md index 46d2e9cb..36ba920a 100644 --- a/.claude/rules/cluster-topology.md +++ b/.claude/rules/cluster-topology.md @@ -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`). - **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". **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". diff --git a/docs/Redundancy.md b/docs/Redundancy.md index ed2ebd41..e47a2ea3 100644 --- a/docs/Redundancy.md +++ b/docs/Redundancy.md @@ -30,7 +30,7 @@ The runtime pieces live in: `RedundancyStateChanged` snapshot, then forwards the result through `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 | |---|---| @@ -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. | | `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`. | +| `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): | 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. | +| 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. | | Stale | 200 | Data stale but ConfigDb reachable. | | 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 > `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 Listing both peers in `SeedNodes` does **not** mean either node can cold-start alone — and neither does any diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Cluster/Redundancy/ServiceLevelCalculator.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Cluster/Redundancy/ServiceLevelCalculator.cs index 0eed29db..2be0b261 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Cluster/Redundancy/ServiceLevelCalculator.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Cluster/Redundancy/ServiceLevelCalculator.cs @@ -2,12 +2,24 @@ using Akka.Cluster; namespace ZB.MOM.WW.OtOpcUa.Cluster.Redundancy; +/// The health facts tiers a node on. +/// This node's Akka cluster member status. +/// Whether the config DB answered (constant true on a DB-less driver node). +/// 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 exists. +/// Whether this node is running behind on config. +/// Whether this node is the driver Primary (oldest Up driver member). +/// This node's OWN view of its OPC UA endpoint: false when the SDK failed +/// to start (IOpcUaEndpointState reports Failed). Defaults to true so a caller with no +/// endpoint fact (a harness, an admin-only node) keeps the previous tiering. lmxopcua#762. public readonly record struct NodeHealthInputs( MemberStatus MemberState, bool DbReachable, bool OpcUaProbeOk, bool Stale, - bool IsDriverPrimary); + bool IsDriverPrimary, + bool EndpointListening = true); /// /// Pure ServiceLevel computation per design §6. Output range 0–255, where higher = "more @@ -16,6 +28,8 @@ public readonly record struct NodeHealthInputs( /// /// Tiering: /// - 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). /// - 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 — @@ -32,6 +46,12 @@ public static class ServiceLevelCalculator if (h.MemberState is not (MemberStatus.Up or MemberStatus.Joining)) 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 { (true, true, false) => 240, diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Host/OpcUa/IOpcUaEndpointState.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Commons/OpcUa/IOpcUaEndpointState.cs similarity index 66% rename from src/Server/ZB.MOM.WW.OtOpcUa.Host/OpcUa/IOpcUaEndpointState.cs rename to src/Core/ZB.MOM.WW.OtOpcUa.Commons/OpcUa/IOpcUaEndpointState.cs index c2377ea1..25646cc3 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Host/OpcUa/IOpcUaEndpointState.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Commons/OpcUa/IOpcUaEndpointState.cs @@ -1,6 +1,6 @@ -namespace ZB.MOM.WW.OtOpcUa.Host.OpcUa; +namespace ZB.MOM.WW.OtOpcUa.Commons.OpcUa; -/// Lifecycle of this node's OPC UA endpoint as drives it. +/// Lifecycle of this node's OPC UA endpoint as the Host's OtOpcUaServerHostedService drives it. public enum OpcUaEndpointStatus { /// The hosted service has not run StartAsync yet. @@ -23,10 +23,18 @@ public enum OpcUaEndpointStatus /// connection-refused from the node the fleet called healthy. /// /// -/// Consumed by OpcUaEndpointHealthCheck (Ready + Active tiers). The redundancy election and the -/// ServiceLevel calculator do not read it yet — they live outside the Host (Cluster / -/// Runtime); feeding them is the Wave 3 follow-up the lane report names. Until then a fused -/// node whose SDK failed is visible (503) but still Primary-eligible. +/// +/// Lives in Commons (not the Host, where it started) so the Runtime can read it. Three consumers: +/// +/// +/// OpcUaEndpointHealthCheck (Host) — Ready tier everywhere, Active tier on a driver-only node. +/// OpcUaPublishActor (Runtime) — a endpoint feeds +/// EndpointListening=false into ServiceLevelCalculator, so the node computes ServiceLevel 0 +/// instead of 250/240 (lmxopcua#762). +/// OpcUaEndpointStepDownActor (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). +/// /// public interface IOpcUaEndpointState { @@ -37,7 +45,8 @@ public interface IOpcUaEndpointState string? FailureReason { get; } } -/// Mutable ; written by the hosted service, read by the health check. +/// Mutable ; written by the hosted service, read by the health check, the +/// ServiceLevel computation and the Primary step-down. public sealed class OpcUaEndpointState : IOpcUaEndpointState { private volatile OpcUaEndpointStatus _status = OpcUaEndpointStatus.NotStarted; diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Health/HealthEndpoints.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Health/HealthEndpoints.cs index 824223dd..0f69625b 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Health/HealthEndpoints.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Health/HealthEndpoints.cs @@ -11,7 +11,7 @@ using ZB.MOM.WW.LocalDb.Replication; using ZB.MOM.WW.OtOpcUa.Cluster; using ZB.MOM.WW.OtOpcUa.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; @@ -45,11 +45,12 @@ public static class HealthEndpoints // 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 // route to it; - // - fused admin+driver: NOT Active, until the ServiceLevel calculator consumes IOpcUaEndpointState - // (follow-up F-C1). Traefik routes the AdminUI on /health/active, so putting the probe there would - // take the operator's only console out of rotation precisely when they need it to diagnose the - // failed endpoint, while the OPC UA failover (ServiceLevel) still would not move. Once F-C1 lands - // the peer takes Primary and the active tier follows it on its own. + // - fused admin+driver: NOT Active. Traefik routes the AdminUI on /health/active, so putting the probe + // there would take the operator's only console out of rotation precisely when they need it to + // diagnose the failed endpoint. The OPC UA failover is handled by the Runtime instead (F-C1, + // lmxopcua#762): the failed node computes ServiceLevel 0 and, when it is the Primary and its peer + // 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. if (hasDriver) { diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Health/OpcUaEndpointHealthCheck.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Health/OpcUaEndpointHealthCheck.cs index e619d7b3..39bd9164 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Health/OpcUaEndpointHealthCheck.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Health/OpcUaEndpointHealthCheck.cs @@ -1,5 +1,5 @@ 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; @@ -12,8 +12,9 @@ namespace ZB.MOM.WW.OtOpcUa.Host.Health; /// /// Registered on driver-role nodes only (); an /// 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 -/// is what Traefik routes the AdminUI on. +/// admin+driver node it is Ready-only, because the Active tier is what Traefik routes the AdminUI on; the +/// Primary moves through the Runtime instead — ServiceLevel 0 plus OpcUaEndpointStepDownActor +/// (lmxopcua#762). /// public sealed class OpcUaEndpointHealthCheck : IHealthCheck { diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/OpcUa/OpcUaPublishActor.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/OpcUa/OpcUaPublishActor.cs index a1667540..be719fc8 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/OpcUa/OpcUaPublishActor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/OpcUa/OpcUaPublishActor.cs @@ -241,6 +241,8 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers /// DbReachable is fed constant true and only snapshot staleness can demote. See /// . private readonly bool _dbLess; + /// This node's OPC UA endpoint lifecycle fact (lmxopcua#762). Null = no fact (harness) = listening. + private readonly IOpcUaEndpointState? _endpointState; private readonly TimeSpan _staleWindow; private readonly TimeSpan _probeFreshnessWindow; private readonly TimeSpan _healthTickInterval; @@ -306,6 +308,9 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers /// ServiceLevel DB-reachability input is retired (DbReachable treated constant-true, only /// staleness demotes). No is wired on such a node. /// The optional sink that receives the API keys and ACL grants carried by each applied artifact. + /// This node's OPC UA endpoint lifecycle fact (lmxopcua#762). A + /// endpoint computes ServiceLevel 0. Null (harnesses, a node with no + /// hosted SDK) is treated as listening. /// The configured for creating this actor, pinned to the /// dispatcher. public static Props Props( @@ -320,7 +325,8 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers TimeSpan? healthTickInterval = null, bool dbLess = false, IDataPlaneAuthoritySink? dataPlaneAuthority = null, - Func? utcNow = null) => + Func? utcNow = null, + IOpcUaEndpointState? endpointState = null) => Akka.Actor.Props.Create(() => new OpcUaPublishActor( sink ?? NullOpcUaAddressSpaceSink.Instance, serviceLevel ?? NullServiceLevelPublisher.Instance, @@ -334,7 +340,8 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers healthTickInterval, dbLess, dataPlaneAuthority, - utcNow)).WithDispatcher(DispatcherId); + utcNow, + endpointState)).WithDispatcher(DispatcherId); /// 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. @@ -359,6 +366,9 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers /// ServiceLevel DB-reachability input is retired (DbReachable treated constant-true, only /// staleness demotes). No is wired on such a node. /// The optional sink that receives the API keys and ACL grants carried by each applied artifact. + /// This node's OPC UA endpoint lifecycle fact (lmxopcua#762). A + /// endpoint computes ServiceLevel 0. Null (harnesses, a node with no + /// hosted SDK) is treated as listening. /// The configured for creating this actor in tests, without dispatcher /// pinning or the DPS subscribe. public static Props PropsForTests( @@ -374,7 +384,8 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers TimeSpan? healthTickInterval = null, bool dbLess = false, IDataPlaneAuthoritySink? dataPlaneAuthority = null, - Func? utcNow = null) => + Func? utcNow = null, + IOpcUaEndpointState? endpointState = null) => Akka.Actor.Props.Create(() => new OpcUaPublishActor( sink ?? NullOpcUaAddressSpaceSink.Instance, serviceLevel ?? NullServiceLevelPublisher.Instance, @@ -388,7 +399,8 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers healthTickInterval, dbLess, dataPlaneAuthority, - utcNow)); + utcNow, + endpointState)); /// Test-only Props exposing the deferred-write bound and the plan-step seams. 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). Akka.Actor.Props.CreateBy(new TestProducer(() => new OpcUaPublishActor( sink, NullServiceLevelPublisher.Instance, false, localNode, null, applier, null, null, null, null, - false, dataPlaneAuthority, null, deferredWriteCapacity, hooks))); + false, dataPlaneAuthority, null, deferredWriteCapacity, hooks, endpointState: null))); /// Produces the actor through its internal (seam-carrying) constructor, for . private sealed class TestProducer(Func create) : IIndirectActorProducer @@ -434,6 +446,7 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers /// ServiceLevel DB-reachability input is retired (DbReachable treated constant-true, only /// staleness demotes). No is wired on such a node. /// The optional sink that receives the API keys and ACL grants carried by each applied artifact. + /// This node's OPC UA endpoint lifecycle fact (lmxopcua#762); null is treated as listening. public OpcUaPublishActor( IOpcUaAddressSpaceSink sink, IServiceLevelPublisher serviceLevel, @@ -447,15 +460,18 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers TimeSpan? healthTickInterval = null, bool dbLess = false, IDataPlaneAuthoritySink? dataPlaneAuthority = null, - Func? utcNow = null) + Func? utcNow = null, + IOpcUaEndpointState? endpointState = null) : 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) { } /// The full constructor: also the deferred-write bound and the test seams (internal — tests only). /// The bound on distinct nodes whose writes are held for replay while a rebuild plan is in flight. /// Test seams for the off-thread plan step; null in production. + /// This node's OPC UA endpoint lifecycle fact (lmxopcua#762); null is treated as listening. internal OpcUaPublishActor( IOpcUaAddressSpaceSink sink, IServiceLevelPublisher serviceLevel, @@ -471,7 +487,8 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers IDataPlaneAuthoritySink? dataPlaneAuthority, Func? utcNow, int deferredWriteCapacity, - TestHooks? hooks) + TestHooks? hooks, + IOpcUaEndpointState? endpointState) { _deferred = new DeferredWrites(Math.Max(1, deferredWriteCapacity)); _hooks = hooks; @@ -486,6 +503,7 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers _applier = applier; _dbHealthProbe = dbHealthProbe; _dbLess = dbLess; + _endpointState = endpointState; _staleWindow = staleWindow ?? TimeSpan.FromSeconds(30); _probeFreshnessWindow = probeFreshnessWindow ?? TimeSpan.FromSeconds(30); _healthTickInterval = healthTickInterval ?? TimeSpan.FromSeconds(5); @@ -1536,6 +1554,15 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers return verdict.Ok; } + /// The OPC UA endpoint input for the calculator (lmxopcua#762): false only when this node's own + /// SDK start is recorded as . + /// and 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 cannot see a failed start + /// when another process holds the port, because its connect then succeeds against that process. + private bool EndpointListening() => _endpointState?.Status != OpcUaEndpointStatus.Failed; + /// /// Computes the local OPC UA ServiceLevel and routes it through /// (the dedup/publish/metric handler). The full path is @@ -1575,7 +1602,8 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers DbReachable: true, OpcUaProbeOk: OpcUaProbeOk(), Stale: (utcNow - entry.AsOfUtc) > _staleWindow, - IsDriverPrimary: entry.IsDriverPrimary); + IsDriverPrimary: entry.IsDriverPrimary, + EndpointListening: EndpointListening()); Self.Tell(new ServiceLevelChanged(ServiceLevelCalculator.Compute(dbLessInputs))); return; } @@ -1585,7 +1613,9 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers // is the bootstrap value until the first DbHealthStatus lands. 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; } @@ -1597,7 +1627,8 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers Stale: !_lastDbHealth.Reachable || (now - _lastDbHealth.AsOfUtc) > _staleWindow || (now - entry.AsOfUtc) > _staleWindow, - IsDriverPrimary: entry.IsDriverPrimary); + IsDriverPrimary: entry.IsDriverPrimary, + EndpointListening: EndpointListening()); Self.Tell(new ServiceLevelChanged(ServiceLevelCalculator.Compute(inputs))); } diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Redundancy/OpcUaEndpointStepDownActor.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Redundancy/OpcUaEndpointStepDownActor.cs new file mode 100644 index 00000000..4095a50d --- /dev/null +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Redundancy/OpcUaEndpointStepDownActor.cs @@ -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; + +/// +/// Steps this node down from driver Primary while its own OPC UA endpoint is not listening (lmxopcua#762). +/// +/// +/// +/// Why. A fused admin+driver node whose SDK failed to start (port 4840 taken, PKI unwritable, a bad +/// OpcUa:* 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. +/// +/// +/// How. The Primary is the oldest Up driver member and must stay exactly where +/// ClusterSingletonManager puts the singletons (see RedundancyStateActor.SelectDriverPrimary), +/// 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 leaves its mesh, the same mechanism as an operator's manual +/// failover: it hands its singletons to the peer, CoordinatedShutdown runs, +/// ActorSystemTerminationWatchdog 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 +/// ServiceLevelCalculator). The peer is now the oldest Up driver member, so it is Primary, hosts the +/// singletons, advertises 250, and the /health/active tier follows it. +/// +/// +/// Guards — every one must hold, checked on each tick: +/// +/// +/// this node's endpoint is (not merely +/// or , which are boot +/// and shutdown windows); +/// this node IS the driver Primary, judged by — the same +/// primitive the election delegates to, so the two cannot disagree; +/// another Up driver member has RECENTLY reported, on , 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. +/// +/// +/// Every node also publishes its own on each tick, which is how its peer learns +/// whether it can take over. The topic is separate from redundancy-state so the half-dozen actors that +/// subscribe there are not handed a message type they do not handle. +/// +/// +public sealed class OpcUaEndpointStepDownActor : ReceiveActor, IWithTimers +{ + /// The DistributedPubSub topic every driver node publishes its on. + public const string Topic = "opcua-endpoint-state"; + + /// Default period between self-reports and step-down evaluations. + public static readonly TimeSpan DefaultCheckInterval = TimeSpan.FromSeconds(5); + + /// Default age beyond which a peer's report is ignored (it went away or stopped reporting). + public static readonly TimeSpan DefaultReportFreshness = TimeSpan.FromSeconds(20); + + /// One node's statement about its own OPC UA endpoint. + /// The reporting node, as host:port of its Akka address. + /// Whether the node's SDK is started and serving. + public sealed record EndpointReport(NodeId Node, bool Listening); + + /// Periodic self-tick. + public sealed class Tick + { + /// The singleton tick instance. + public static readonly Tick Instance = new(); + private Tick() { } + } + + private readonly IOpcUaEndpointState _state; + private readonly TimeSpan _interval; + private readonly TimeSpan _freshness; + private readonly Func _utcNow; + private readonly Akka.Cluster.Cluster _cluster = Akka.Cluster.Cluster.Get(Context.System); + private readonly ILoggingAdapter _log = Context.GetLogger(); + private readonly Dictionary _peerReports = new(); + private readonly NodeId _self; + private bool _leaving; + private string? _lastHoldReason; + + /// Gets or sets the timer scheduler. + public ITimerScheduler Timers { get; set; } = null!; + + /// Creates the actor's props. + /// This node's OPC UA endpoint lifecycle fact. + /// Period between self-reports and evaluations; defaults to . + /// Peer-report age limit; defaults to . + /// Clock; defaults to . + /// The props. + public static Props Props( + IOpcUaEndpointState state, + TimeSpan? checkInterval = null, + TimeSpan? reportFreshness = null, + Func? utcNow = null) => + Akka.Actor.Props.Create(() => new OpcUaEndpointStepDownActor(state, checkInterval, reportFreshness, utcNow)); + + /// Initializes a new instance of the class. + /// This node's OPC UA endpoint lifecycle fact. + /// Period between self-reports and evaluations. + /// Peer-report age limit. + /// Clock. + public OpcUaEndpointStepDownActor( + IOpcUaEndpointState state, + TimeSpan? checkInterval, + TimeSpan? reportFreshness, + Func? utcNow) + { + _state = state ?? throw new ArgumentNullException(nameof(state)); + _interval = checkInterval ?? DefaultCheckInterval; + _freshness = reportFreshness ?? DefaultReportFreshness; + _utcNow = utcNow ?? (() => DateTime.UtcNow); + _self = ToNodeId(_cluster.SelfAddress); + + Receive(_ => OnTick()); + Receive(OnReport); + Receive(_ => { }); + } + + /// + /// The canonical host:port id of an Akka address — the format RedundancyStateActor.ToNodeId and + /// ClusterRoleInfo use. Derived from the member address on both sides (reporter and reader), so the two + /// always agree. + /// + /// The member address. + /// The node id. + public static NodeId ToNodeId(Address address) => + NodeId.Parse($"{address.Host ?? string.Empty}:{address.Port ?? 0}"); + + /// + 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); + } +} diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ServiceCollectionExtensions.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ServiceCollectionExtensions.cs index ea327b42..6e5c5c4e 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ServiceCollectionExtensions.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ServiceCollectionExtensions.cs @@ -44,6 +44,7 @@ public static class ServiceCollectionExtensions public const string DependencyMuxActorName = "dependency-mux"; public const string OpcUaPublishActorName = "opcua-publish"; public const string PeerProbeSupervisorName = "peer-probe-supervisor"; + public const string OpcUaEndpointStepDownActorName = "opcua-endpoint-step-down"; public const string ContinuousHistorizationRecorderActorName = "continuous-historization-recorder"; /// @@ -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; // the actor treats null as "nothing consumes authorization inputs" and skips the decode. var dataPlaneAuthority = resolver.GetService(); + // 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(); var publishActor = system.ActorOf( OpcUaPublishActor.Props( sink: addressSpaceSink, @@ -564,10 +569,19 @@ public static class ServiceCollectionExtensions applier: applier, dbHealthProbe: dbHealth, dbLess: dbFactory is null, - dataPlaneAuthority: dataPlaneAuthority), + dataPlaneAuthority: dataPlaneAuthority, + endpointState: endpointState), OpcUaPublishActorName); registry.Register(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 // 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. diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ZB.MOM.WW.OtOpcUa.Runtime.csproj b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ZB.MOM.WW.OtOpcUa.Runtime.csproj index d45ea40d..aca6dd21 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ZB.MOM.WW.OtOpcUa.Runtime.csproj +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ZB.MOM.WW.OtOpcUa.Runtime.csproj @@ -23,6 +23,10 @@ + + diff --git a/tests/Core/ZB.MOM.WW.OtOpcUa.Cluster.Tests/ServiceLevelCalculatorTests.cs b/tests/Core/ZB.MOM.WW.OtOpcUa.Cluster.Tests/ServiceLevelCalculatorTests.cs index 7c9edbc4..c90a7e19 100644 --- a/tests/Core/ZB.MOM.WW.OtOpcUa.Cluster.Tests/ServiceLevelCalculatorTests.cs +++ b/tests/Core/ZB.MOM.WW.OtOpcUa.Cluster.Tests/ServiceLevelCalculatorTests.cs @@ -88,4 +88,32 @@ public sealed class ServiceLevelCalculatorTests DbReachable: true, OpcUaProbeOk: true, Stale: false, IsDriverPrimary: true)); ((int)sl).ShouldBeLessThanOrEqualTo(255); } + + /// + /// 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. + /// + [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); + } + + /// The endpoint input defaults to listening, so callers with no endpoint fact keep the old tiering. + [Fact] + public void Endpoint_listening_defaults_to_true() + { + new NodeHealthInputs(MemberStatus.Up, true, true, false, true).EndpointListening.ShouldBeTrue(); + } } diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/OpcUa/OpcUaPublishActorEndpointStateTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/OpcUa/OpcUaPublishActorEndpointStateTests.cs new file mode 100644 index 00000000..9e51513d --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/OpcUa/OpcUaPublishActorEndpointStateTests.cs @@ -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; + +/// +/// 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). +/// +public sealed class OpcUaPublishActorEndpointStateTests : RuntimeActorTestBase +{ + /// DB-less driver node (the survive-alone posture): Primary would be 250, a failed endpoint is 0. + [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); + } + + /// DB-backed node on the full calculator path (DB reachable, fresh): same rule. + [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); + } + + /// The legacy role-only seam (no DB probe wired yet) must not hand a failed endpoint 240 either. + [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); + } + + /// + /// Boot () and shutdown () + /// are not faults: only a recorded start FAILURE demotes. + /// + [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); + } + + /// + /// 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. + /// + [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 _levels = new(); + public byte[] Levels => _levels.ToArray(); + public void Publish(byte serviceLevel) => _levels.Enqueue(serviceLevel); + } +} diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Redundancy/OpcUaEndpointStepDownActorTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Redundancy/OpcUaEndpointStepDownActorTests.cs new file mode 100644 index 00000000..2ac05597 --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Redundancy/OpcUaEndpointStepDownActorTests.cs @@ -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; + +/// +/// 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. +/// +/// +/// Node A is a second that forms the cluster first, so it is the OLDEST Up driver +/// member — the node the election () +/// names Primary. This class's own is node B, which joins second; it survives +/// every test, so the base class disposes it normally. +/// +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); + + /// How long a "does NOT leave" assertion watches — many ticks, and well past pub-sub convergence. + 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(); + } + + /// + protected override void Dispose(bool disposing) + { + if (disposing) _nodeA.Dispose(); + base.Dispose(disposing); + } + + /// + /// 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. + /// + [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); + } + + /// A failed endpoint on the SECONDARY changes nothing: there is no role to hand over. + [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(); + } + + /// A healthy Primary never leaves, even with a listening peer (positive control for the trigger). + [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); +}