diff --git a/docs/AlarmHistorian.md b/docs/AlarmHistorian.md index 991d01b2..7f29303f 100644 --- a/docs/AlarmHistorian.md +++ b/docs/AlarmHistorian.md @@ -103,6 +103,17 @@ Two postures are deliberate: a key with no dial and no listener (e.g. a fleet-wide secret on an unpaired node) captures changes but shares them with nobody, and gating it would park its events behind a Primary that does not hold them (review R1 m3). +- **A gated node stands down only for the peer that holds its rows — and BOTH + halves of a pair know it.** The dialling half identifies its peer by the host + of `LocalDb:Replication:PeerAddress`. The passive half (`ApiKey` + + `SyncListenPort`, no `PeerAddress`) identifies it as its **per-cluster mesh + partner** (the other member of its `cluster-{ClusterId}` role). Before review + R6 MINOR-4 the passive half knew no peer at all, so as Secondary it kept + draining beside its Primary partner and the historian received every alarm + event twice. ⚠️ A passive half with **no** cluster-scoped role (a legacy, + pre-Phase-6 node) still cannot identify its partner and keeps draining — it + cannot prove which Primary holds its rows, and deferring to the wrong one + would strand them. - **A gate that throws means "not now", never permission.** Draining on a gate fault would put a pair into dual delivery exactly when its redundancy signal is sick. diff --git a/docs/Redundancy.md b/docs/Redundancy.md index c8f88d59..d031b520 100644 --- a/docs/Redundancy.md +++ b/docs/Redundancy.md @@ -532,7 +532,7 @@ Under warm/hot redundancy both cluster nodes run `ScriptedAlarmHostActor` and ev - **`alerts` topic emission** — `ScriptedAlarmHostActor` and `DriverHostActor.ForwardNativeAlarm` subscribe to the `redundancy-state` DPS topic and cache the local node's `RedundancyRole`, then gate the cluster `alerts` publish through `PrimaryGatePolicy` (table above). The OPC UA condition-node write and inbound ack/shelve command processing remain **ungated** on both nodes so the secondary is always ready to serve clients after a failover. - **`HistorianAdapterActor` historization** — likewise Primary-gated so alarm historization is exactly-once across all alarm sources. The actor subscribes to the `alerts` DPS topic and translates each `AlarmTransitionEvent` → `AlarmHistorianEvent` before enqueuing it on the sink; scripted alarms therefore historize exactly once regardless of cluster size. -- **The alarm sink's DRAIN** — a second, independent gate on the same decision. The enqueue gate above is not sufficient any more: since Phase 2 the store-and-forward buffer lives in the replicated LocalDb, so the Secondary holds a full copy of the Primary's queued rows and an ungated drain there would re-deliver every event continuously. `LocalDbStoreAndForwardSink` takes a `drainGate` fed from `IRedundancyRoleView` — a singleton `DriverHostActor` publishes its `PrimaryGatePolicy` verdict to, because the drain runs on a timer and cannot receive `RedundancyStateChanged` itself. A gated-off node reports `HistorianDrainState.NotPrimary`. The gate applies only to a node whose queue a peer actually holds — `LocalDbReplicationPolicy.SharesRowsWithPeer`: `LocalDb:Replication:ApiKey` AND (`PeerAddress` OR `LocalDb:SyncListenPort` > 0); any other node always drains (review R1 m3). Delivery is **at-least-once across a failover** by design: rows the old Primary delivered just before dying may not have replicated their deletes yet. See [AlarmHistorian.md](AlarmHistorian.md). +- **The alarm sink's DRAIN** — a second, independent gate on the same decision. The enqueue gate above is not sufficient any more: since Phase 2 the store-and-forward buffer lives in the replicated LocalDb, so the Secondary holds a full copy of the Primary's queued rows and an ungated drain there would re-deliver every event continuously. `LocalDbStoreAndForwardSink` takes a `drainGate` fed from `IRedundancyRoleView` — a singleton `DriverHostActor` publishes its `PrimaryGatePolicy` verdict to, because the drain runs on a timer and cannot receive `RedundancyStateChanged` itself. A gated-off node reports `HistorianDrainState.NotPrimary`. The gate applies only to a node whose queue a peer actually holds — `LocalDbReplicationPolicy.SharesRowsWithPeer`: `LocalDb:Replication:ApiKey` AND (`PeerAddress` OR `LocalDb:SyncListenPort` > 0); any other node always drains (review R1 m3). Within a pair BOTH halves stand down for their partner: the dialling half knows it by the `PeerAddress` host, the passive half (no `PeerAddress`) as its per-cluster mesh partner (review R6 MINOR-4 — before that the passive half never stood down, so both halves drained and every event reached the historian twice; a passive half with no `cluster-{ClusterId}` role still cannot identify its partner and keeps draining). Delivery is **at-least-once across a failover** by design: rows the old Primary delivered just before dying may not have replicated their deletes yet. See [AlarmHistorian.md](AlarmHistorian.md). - **Continuous-historization value CAPTURE** (Gitea #537) — a third gate on the same role rule, and the one that gates the *opposite* half of its pipeline. Both pair nodes serve the same address space and see diff --git a/docs/operations/2026-07-20-localdb-pair-replication.md b/docs/operations/2026-07-20-localdb-pair-replication.md index b61a94ed..5db75b75 100644 --- a/docs/operations/2026-07-20-localdb-pair-replication.md +++ b/docs/operations/2026-07-20-localdb-pair-replication.md @@ -94,7 +94,10 @@ Replication has two knobs, both under `LocalDb`: > the passive node has no `PeerAddress`; `SyncListenPort` is deliberately not part of it, since a > port with no key is fail-closed and syncs with nobody). The alarm-historian drain gate uses the > stricter `ApiKey` AND (`PeerAddress` OR `SyncListenPort > 0`) — so set the key **and** the port/peer -> together, or the node keeps draining its own queue. Consequences at enablement time: +> together, or the node keeps draining its own queue. Once gated, the passive half stands down for +> its **per-cluster mesh partner** (it has no `PeerAddress` to name one) — so it must carry the pair's +> `cluster-{ClusterId}` role, or as Secondary it keeps draining beside its Primary and the historian +> gets every alarm event twice (review R6 MINOR-4). Consequences at enablement time: > > - A configured node registers with **`baselineExistingRows: true`**, so the rows **already in the > file** are seeded into the row-version ledger at the LWW floor (HLC `0`) and actually travel. diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs index 5c72c36d..33ed440a 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs @@ -472,6 +472,14 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS /// private readonly string? _replicationPeerHost; + /// + /// Review R6 MINOR-4: identifies the alarm-queue-sharing peer when is + /// unknown — the passive (listening) half of a pair, which replicates the same rows but dials nobody. Before + /// this the passive half had no peer to defer to, so as Secondary it kept draining alongside its Primary + /// partner and every alarm event reached the historian twice. Null ⇒ no peer known ⇒ never stand down. + /// + private readonly Func? _sharesAlarmQueueWithLocal; + /// Debounces the S5 "snapshot omitted this node" Warning to once per process — a persistent /// identity mismatch (03/S5) would otherwise log on every snapshot. private bool _warnedSnapshotMissingLocalNode; @@ -617,7 +625,8 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS Func? sharesHistorizationScopeWithLocal = null, TimeSpan? applyWedgeTimeout = null, TimeSpan? reprovisionBackoff = null, - TimeSpan? commsRecoveryAuditDelay = null) => + TimeSpan? commsRecoveryAuditDelay = null, + Func? sharesAlarmQueueWithLocal = null) => // WARNING: this forwarding list is POSITIONAL, and Props.Create compiles it into an // expression tree. Six IActorRef? parameters and several interface-typed ones mean a // mis-ordered argument is usually type-compatible and therefore compiles clean, then binds @@ -629,7 +638,8 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS loggerFactory, scriptRootLogger, scriptedAlarmHostOverride, invokerFactory, driverMemberCountProvider, deploymentArtifactCache, redundancyRoleView, replicationPeerHost, fetchAndCacheMode, artifactFetcher, alarmStateStore, telemetryHub, primaryRoleSource, historizationRoleView, - sharesHistorizationScopeWithLocal, applyWedgeTimeout, reprovisionBackoff, commsRecoveryAuditDelay)); + sharesHistorizationScopeWithLocal, applyWedgeTimeout, reprovisionBackoff, commsRecoveryAuditDelay, + sharesAlarmQueueWithLocal)); /// Initializes a new DriverHostActor with the specified dependencies. /// Database context factory for configuration database access. @@ -691,6 +701,12 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS /// Archreview 03/S15: how long after a driver reconnects the /// recovery audit waits before reporting refs still latched BadNoCommunication. Defaults to /// ; a test shortens it for the same reason. + /// Review R6 MINOR-4: tests whether a node named in a redundancy + /// snapshot holds this node's replicated alarm queue, for a node that knows no + /// — the PASSIVE half of a LocalDb pair (ApiKey + SyncListenPort, no + /// PeerAddress), whose partner is its per-cluster mesh partner. Ignored when + /// is set. Null ⇒ no such peer is known and this node never stands its + /// drain down. public DriverHostActor( IDbContextFactory? dbFactory, CommonsNodeId localNode, @@ -720,8 +736,10 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS Func? sharesHistorizationScopeWithLocal = null, TimeSpan? applyWedgeTimeout = null, TimeSpan? reprovisionBackoff = null, - TimeSpan? commsRecoveryAuditDelay = null) + TimeSpan? commsRecoveryAuditDelay = null, + Func? sharesAlarmQueueWithLocal = null) { + _sharesAlarmQueueWithLocal = sharesAlarmQueueWithLocal; _applyWedgeTimeout = applyWedgeTimeout ?? DefaultApplyWedgeTimeout; _reprovisionBackoff = reprovisionBackoff ?? DefaultReprovisionBackoff; _commsRecoveryAuditDelay = commsRecoveryAuditDelay ?? DefaultCommsRecoveryAuditDelay; @@ -2022,6 +2040,14 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS } } + /// Whether holds this node's replicated alarm queue: the PeerAddress host when + /// this node dials, else the passive-half predicate (review R6 MINOR-4). This node itself never counts. + private bool SharesAlarmQueue(CommonsNodeId nodeId) => + nodeId != _localNode + && (_replicationPeerHost is not null + ? HostOf(nodeId) == _replicationPeerHost + : _sharesAlarmQueueWithLocal?.Invoke(nodeId) == true); + /// Host part of a host:port node id. private static string HostOf(NodeId nodeId) { @@ -2071,10 +2097,10 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS // Defer ONLY to the node that holds a copy of our rows. The redundancy election is // cluster-wide and the queue is pair-local, so "somebody is Primary" is not a licence to stop // draining — that Primary is usually in another pair entirely, and cannot deliver our events. - var peerIsPrimary = - _replicationPeerHost is not null - && msg.Nodes.Any(n => n.Role == RedundancyRole.Primary - && HostOf(n.NodeId) == _replicationPeerHost); + // + // The dialling half knows its peer by PeerAddress host. The passive half (review R6 MINOR-4) knows it + // only as its mesh partner — the predicate — and must be gated too, or both halves drain the same rows. + var peerIsPrimary = msg.Nodes.Any(n => n.Role == RedundancyRole.Primary && SharesAlarmQueue(n.NodeId)); _redundancyRoleView?.Publish( PrimaryGatePolicy.ShouldDrainAlarmHistory(_localRole, queueSharingPeerIsPrimary: peerIsPrimary)); diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ServiceCollectionExtensions.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ServiceCollectionExtensions.cs index c50deff9..516be128 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ServiceCollectionExtensions.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ServiceCollectionExtensions.cs @@ -279,6 +279,33 @@ public static class ServiceCollectionExtensions return services; } + /// + /// Review R6 MINOR-4 — "does this snapshot node hold my replicated alarm queue?" for a node that does not + /// dial its peer. Both halves of a LocalDb pair hold the same rows, but only the dialling half knows its + /// partner (LocalDb:Replication:PeerAddress); the passive half (ApiKey + SyncListenPort) + /// knew nobody, so as Secondary it never stood its drain down and both nodes delivered every alarm event. + /// Its partner is its per-cluster mesh partner, which a cluster-scoped role identifies. + /// + /// The host configuration (the LocalDb replication keys). + /// This node's cluster role info. + /// The dialled peer's host, when this node dials one. + /// + /// null when the node dials its peer (the host match covers it), when it shares its queue with no + /// peer ( false — it must never stand down), or + /// when it carries no cluster-scoped role (a legacy node cannot prove which node is its partner, and + /// standing down for the wrong Primary would strand its rows); otherwise "is a member of my own mesh other + /// than me". + /// + internal static Func? BuildAlarmQueuePeerPredicate( + IConfiguration? configuration, IClusterRoleInfo? roleInfo, string? replicationPeerHost) + { + if (replicationPeerHost is not null || configuration is null || roleInfo is null) return null; + if (!LocalDbReplicationPolicy.SharesRowsWithPeer(configuration)) return null; + if (roleInfo.ClusterRole is not { } meshRole) return null; + var local = roleInfo.LocalNode; + return nodeId => nodeId != local && roleInfo.MembersWithRole(meshRole).Contains(nodeId); + } + /// /// Spawns the per-node driver-role actors on the host's : /// (one per node), @@ -361,7 +388,8 @@ public static class ServiceCollectionExtensions var redundancyRoleView = resolver.GetService(); // Host of this node's LocalDb replication partner: the ONLY node that holds a copy of this // node's alarm queue, and so the only node it may stand down in favour of. Null when this - // node dials nobody, which correctly means "never stand down". + // node dials nobody — unpaired (never stand down), or the PASSIVE half of a pair, which is + // told its partner through BuildAlarmQueuePeerPredicate instead (review R6 MINOR-4). var replicationPeerHost = Uri.TryCreate( resolver.GetService()?["LocalDb:Replication:PeerAddress"], @@ -592,7 +620,11 @@ public static class ServiceCollectionExtensions // Gitea #537: where this actor publishes the continuous-historization CAPTURE // verdict, and how it decides which peer is worth standing down for. historizationRoleView: resolver.GetService(), - sharesHistorizationScopeWithLocal: historizationScopePeer), + sharesHistorizationScopeWithLocal: historizationScopePeer, + // Review R6 MINOR-4: the passive half of a LocalDb pair has no PeerAddress, so it is told + // its queue-sharing peer as its mesh partner instead — otherwise it never stands down. + sharesAlarmQueueWithLocal: BuildAlarmQueuePeerPredicate( + resolver.GetService(), roleInfo, replicationPeerHost)), DriverHostActorName); registry.Register(driverHost); diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorRoleViewTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorRoleViewTests.cs index 0e0ec0ce..92d376e9 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorRoleViewTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorRoleViewTests.cs @@ -135,6 +135,43 @@ public sealed class DriverHostActorRoleViewTests : RuntimeActorTestBase AwaitAssert(() => view.Published.ShouldBe([true, false]), duration: Timeout); } + /// + /// Review R6 MINOR-4 — the PASSIVE half of a pair dials nobody (no PeerAddress, so no peer host) but + /// holds the same replicated rows, so as Secondary it must stand down for its Primary partner too, which it + /// identifies through the queue-sharing predicate (its mesh partner). Before the fix it published + /// true here and both halves drained. A Primary the predicate does NOT name is the control: the + /// passive half keeps draining, so the deny comes from identifying the partner, not from being Secondary. + /// + [Fact] + public void A_passive_secondary_closes_the_view_for_its_mesh_partner_and_only_for_it() + { + var partner = NodeId.Parse("site-a-1:4053"); + foreach (var (primary, expected) in new[] { (partner, false), (NodeId.Parse("site-b-1:4053"), true) }) + { + var view = new RecordingRoleView(); + var host = Sys.ActorOf(DriverHostActor.Props( + NewInMemoryDbFactory(), + TestNode, + ackRouter: null, + driverMemberCountProvider: () => 2, + redundancyRoleView: view, + replicationPeerHost: null, + sharesAlarmQueueWithLocal: n => n == partner)); + + host.Tell(new RedundancyStateChanged( + [ + new NodeRedundancyState(TestNode, RedundancyRole.Secondary, + IsClusterLeader: false, IsDriverPrimary: false, AsOfUtc: DateTime.UtcNow), + new NodeRedundancyState(primary, RedundancyRole.Primary, + IsClusterLeader: true, IsDriverPrimary: true, AsOfUtc: DateTime.UtcNow), + ], + CorrelationId.NewId())); + + AwaitAssert(() => view.Published.ShouldBe([expected]), duration: Timeout); + Sys.Stop(host); + } + } + // ---------------- helpers ---------------- /// diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Historian/AlarmQueuePeerPredicateTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Historian/AlarmQueuePeerPredicateTests.cs new file mode 100644 index 00000000..6f087e36 --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Historian/AlarmQueuePeerPredicateTests.cs @@ -0,0 +1,83 @@ +using Microsoft.Extensions.Configuration; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Commons.Interfaces; +using ZB.MOM.WW.OtOpcUa.Commons.Types; + +namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Historian; + +/// +/// Review R6 MINOR-4 — how the PASSIVE half of a LocalDb pair (ApiKey + SyncListenPort, no +/// PeerAddress) learns which node holds its replicated alarm queue: its per-cluster mesh partner. +/// Before this it had no peer at all, so as Secondary it kept draining beside its Primary partner and the +/// historian received every alarm event twice. +/// +public sealed class AlarmQueuePeerPredicateTests +{ + private static readonly NodeId Self = NodeId.Parse("site-a-2:4053"); + private static readonly NodeId Partner = NodeId.Parse("site-a-1:4053"); + private static readonly NodeId OtherPair = NodeId.Parse("site-b-1:4053"); + + /// The passive half (key + listener) with a mesh role: its mesh partner is its queue peer, itself and + /// a node from another mesh are not. + [Fact] + public void The_passive_half_identifies_its_mesh_partner() + { + var predicate = ServiceCollectionExtensions.BuildAlarmQueuePeerPredicate( + Config(apiKey: "k", syncListenPort: "4057"), new MeshRoleInfo("cluster-SITE-A", Self, Partner), replicationPeerHost: null); + + predicate.ShouldNotBeNull(); + predicate(Partner).ShouldBeTrue(); + predicate(Self).ShouldBeFalse(); + predicate(OtherPair).ShouldBeFalse(); + } + + /// The dialling half is matched by its PeerAddress host instead — no predicate. + [Fact] + public void The_dialling_half_needs_no_predicate() => + ServiceCollectionExtensions.BuildAlarmQueuePeerPredicate( + Config(apiKey: "k", peerAddress: "http://site-a-2:4057"), new MeshRoleInfo("cluster-SITE-A", Self, Partner), + replicationPeerHost: "site-a-2").ShouldBeNull(); + + /// A node sharing its queue with nobody must never stand down, whatever its mesh says. + [Theory] + [InlineData(null, "4057")] // listener without the key — fail-closed, shares nothing + [InlineData("k", null)] // key without a path — shares nothing + [InlineData(null, null)] // unreplicated + public void A_node_that_shares_no_rows_gets_no_predicate(string? apiKey, string? syncListenPort) => + ServiceCollectionExtensions.BuildAlarmQueuePeerPredicate( + Config(apiKey: apiKey, syncListenPort: syncListenPort), new MeshRoleInfo("cluster-SITE-A", Self, Partner), + replicationPeerHost: null).ShouldBeNull(); + + /// A legacy node with no cluster-scoped role cannot prove which node is its partner; standing down for + /// the wrong Primary would strand its rows, so it gets none (and keeps draining). + [Fact] + public void A_node_without_a_mesh_role_gets_no_predicate() => + ServiceCollectionExtensions.BuildAlarmQueuePeerPredicate( + Config(apiKey: "k", syncListenPort: "4057"), new MeshRoleInfo(null, Self, Partner), replicationPeerHost: null) + .ShouldBeNull(); + + private static IConfiguration Config(string? apiKey = null, string? peerAddress = null, string? syncListenPort = null) => + new ConfigurationBuilder().AddInMemoryCollection(new Dictionary + { + ["LocalDb:Replication:ApiKey"] = apiKey, + ["LocalDb:Replication:PeerAddress"] = peerAddress, + ["LocalDb:SyncListenPort"] = syncListenPort, + }).Build(); + + private sealed class MeshRoleInfo(string? clusterRole, NodeId self, NodeId partner) : IClusterRoleInfo + { + public NodeId LocalNode => self; + public IReadOnlySet LocalRoles { get; } = new HashSet(["driver"]); + public string? ClusterRole => clusterRole; + public string? ClusterId => clusterRole?["cluster-".Length..]; + public bool HasRole(string role) => LocalRoles.Contains(role) || role == clusterRole; + public IReadOnlyList MembersWithRole(string role) => role == clusterRole ? [self, partner] : []; + public NodeId? RoleLeader(string role) => null; + public event EventHandler? RoleLeaderChanged + { + add { _ = value; } + remove { _ = value; } + } + } +}