fix(runtime): the passive half of a LocalDb pair stands its alarm drain down for its mesh partner (review R6 MINOR-4)

The drain gate's 'is my queue-sharing peer Primary?' used only the host of
LocalDb:Replication:PeerAddress, which the passive half (ApiKey +
SyncListenPort) does not set - so as Secondary it kept draining beside its
Primary partner and every alarm event reached the historian twice.
DriverHostActor takes a sharesAlarmQueueWithLocal predicate used when no
peer host is known; WithOtOpcUaRuntimeActors builds it
(BuildAlarmQueuePeerPredicate) for a SharesRowsWithPeer node with no
PeerAddress as 'a member of my own cluster-{ClusterId} mesh other than
me'. A node sharing no rows, or with no mesh role, gets none and keeps
draining. AlarmHistorian.md / Redundancy.md / pair-replication runbook
updated.

Claude-Session: https://claude.ai/code/session_01Ef2ozxS7QPanGT8uoATbTe

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.

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

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.

src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs

@@ -472,6 +472,14 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
     /// </summary>
     private readonly string? _replicationPeerHost;
 
+    /// <summary>
+    ///     Review R6 MINOR-4: identifies the alarm-queue-sharing peer when <see cref="_replicationPeerHost"/> 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.
+    /// </summary>
+    private readonly Func<CommonsNodeId, bool>? _sharesAlarmQueueWithLocal;
+
     /// <summary>Debounces the S5 "snapshot omitted this node" Warning to once per process — a persistent
     /// identity mismatch (03/S5) would otherwise log on every snapshot.</summary>
     private bool _warnedSnapshotMissingLocalNode;
@@ -617,7 +625,8 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
         Func<CommonsNodeId, bool>? sharesHistorizationScopeWithLocal = null,
         TimeSpan? applyWedgeTimeout = null,
         TimeSpan? reprovisionBackoff = null,
-        TimeSpan? commsRecoveryAuditDelay = null) =>
+        TimeSpan? commsRecoveryAuditDelay = null,
+        Func<CommonsNodeId, bool>? 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));
 
     /// <summary>Initializes a new DriverHostActor with the specified dependencies.</summary>
     /// <param name="dbFactory">Database context factory for configuration database access.</param>
@@ -691,6 +701,12 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
     /// <param name="commsRecoveryAuditDelay">Archreview 03/S15: how long after a driver reconnects the
     /// recovery audit waits before reporting refs still latched <c>BadNoCommunication</c>. Defaults to
     /// <see cref="DefaultCommsRecoveryAuditDelay"/>; a test shortens it for the same reason.</param>
+    /// <param name="sharesAlarmQueueWithLocal">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
+    /// <paramref name="replicationPeerHost"/> — the PASSIVE half of a LocalDb pair (ApiKey + SyncListenPort, no
+    /// PeerAddress), whose partner is its per-cluster mesh partner. Ignored when
+    /// <paramref name="replicationPeerHost"/> is set. Null ⇒ no such peer is known and this node never stands its
+    /// drain down.</param>
     public DriverHostActor(
         IDbContextFactory<OtOpcUaConfigDbContext>? dbFactory,
         CommonsNodeId localNode,
@@ -720,8 +736,10 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
         Func<CommonsNodeId, bool>? sharesHistorizationScopeWithLocal = null,
         TimeSpan? applyWedgeTimeout = null,
         TimeSpan? reprovisionBackoff = null,
-        TimeSpan? commsRecoveryAuditDelay = null)
+        TimeSpan? commsRecoveryAuditDelay = null,
+        Func<CommonsNodeId, bool>? sharesAlarmQueueWithLocal = null)
     {
+        _sharesAlarmQueueWithLocal = sharesAlarmQueueWithLocal;
         _applyWedgeTimeout = applyWedgeTimeout ?? DefaultApplyWedgeTimeout;
         _reprovisionBackoff = reprovisionBackoff ?? DefaultReprovisionBackoff;
         _commsRecoveryAuditDelay = commsRecoveryAuditDelay ?? DefaultCommsRecoveryAuditDelay;
@@ -2022,6 +2040,14 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
         }
     }
 
+    /// <summary>Whether <paramref name="nodeId"/> 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.</summary>
+    private bool SharesAlarmQueue(CommonsNodeId nodeId) =>
+        nodeId != _localNode
+        && (_replicationPeerHost is not null
+            ? HostOf(nodeId) == _replicationPeerHost
+            : _sharesAlarmQueueWithLocal?.Invoke(nodeId) == true);
+
     /// <summary>Host part of a <c>host:port</c> node id.</summary>
     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
+        // The dialling half knows its peer by PeerAddress host. The passive half (review R6 MINOR-4) knows it
-            && msg.Nodes.Any(n => n.Role == RedundancyRole.Primary
+        // only as its mesh partner — the predicate — and must be gated too, or both halves drain the same rows.
-                                  && HostOf(n.NodeId) == _replicationPeerHost);
+        var peerIsPrimary = msg.Nodes.Any(n => n.Role == RedundancyRole.Primary && SharesAlarmQueue(n.NodeId));
 
         _redundancyRoleView?.Publish(
             PrimaryGatePolicy.ShouldDrainAlarmHistory(_localRole, queueSharingPeerIsPrimary: peerIsPrimary));

src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ServiceCollectionExtensions.cs

@@ -279,6 +279,33 @@ public static class ServiceCollectionExtensions
         return services;
     }
 
+    /// <summary>
+    ///     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 (<c>LocalDb:Replication:PeerAddress</c>); the passive half (<c>ApiKey</c> + <c>SyncListenPort</c>)
+    ///     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.
+    /// </summary>
+    /// <param name="configuration">The host configuration (the LocalDb replication keys).</param>
+    /// <param name="roleInfo">This node's cluster role info.</param>
+    /// <param name="replicationPeerHost">The dialled peer's host, when this node dials one.</param>
+    /// <returns>
+    ///     <c>null</c> when the node dials its peer (the host match covers it), when it shares its queue with no
+    ///     peer (<see cref="LocalDbReplicationPolicy.SharesRowsWithPeer"/> 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".
+    /// </returns>
+    internal static Func<Commons.Types.NodeId, bool>? 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);
+    }
+
     /// <summary>
     /// Spawns the per-node driver-role actors on the host's <see cref="ActorSystem"/>:
     /// <see cref="DriverHostActor"/> (one per node), <see cref="DbHealthProbeActor"/>
@@ -361,7 +388,8 @@ public static class ServiceCollectionExtensions
             var redundancyRoleView = resolver.GetService<IRedundancyRoleView>();
             // 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<IConfiguration>()?["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<IHistorizationRoleView>(),
-                    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<IConfiguration>(), roleInfo, replicationPeerHost)),
                 DriverHostActorName);
             registry.Register<DriverHostActorKey>(driverHost);
 

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);
     }
 
+    /// <summary>
+    ///     Review R6 MINOR-4 — the PASSIVE half of a pair dials nobody (no <c>PeerAddress</c>, 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
+    ///     <c>true</c> 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.
+    /// </summary>
+    [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 ----------------
 
     /// <summary>

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;
+
+/// <summary>
+///     Review R6 MINOR-4 — how the PASSIVE half of a LocalDb pair (<c>ApiKey</c> + <c>SyncListenPort</c>, no
+///     <c>PeerAddress</c>) 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.
+/// </summary>
+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");
+
+    /// <summary>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.</summary>
+    [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();
+    }
+
+    /// <summary>The dialling half is matched by its PeerAddress host instead — no predicate.</summary>
+    [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();
+
+    /// <summary>A node sharing its queue with nobody must never stand down, whatever its mesh says.</summary>
+    [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();
+
+    /// <summary>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).</summary>
+    [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<string, string?>
+        {
+            ["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<string> LocalRoles { get; } = new HashSet<string>(["driver"]);
+        public string? ClusterRole => clusterRole;
+        public string? ClusterId => clusterRole?["cluster-".Length..];
+        public bool HasRole(string role) => LocalRoles.Contains(role) || role == clusterRole;
+        public IReadOnlyList<NodeId> MembersWithRole(string role) => role == clusterRole ? [self, partner] : [];
+        public NodeId? RoleLeader(string role) => null;
+        public event EventHandler<RoleLeaderChangedEventArgs>? RoleLeaderChanged
+        {
+            add { _ = value; }
+            remove { _ = value; }
+        }
+    }
+}