refactor(localdb): one replication predicate shared by table registration and the alarm drain gate (#603)

The drain gate counted SyncListenPort; a port with no ApiKey shares nothing (fail-closed
passive endpoint), and an ApiKey-only passive node shares its queue but was never gated.

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

src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/LocalDbRegistration.cs

@@ -40,7 +40,7 @@ public static class LocalDbRegistration
     public const string LocalDbSectionPath = "LocalDb";
 
     /// <summary>Configuration section holding <c>ReplicationOptions</c>.</summary>
-    public const string ReplicationSectionPath = "LocalDb:Replication";
+    public const string ReplicationSectionPath = Runtime.LocalDbReplicationPolicy.ReplicationSectionPath;
 
     /// <summary>
     ///     Configuration key carrying the dedicated h2c sync listener's port. Zero or absent means

src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/LocalDbSetup.cs

@@ -170,12 +170,9 @@ public static class LocalDbSetup
     /// <returns><c>true</c> when this node is configured to replicate.</returns>
     public static bool ReplicationIsConfigured(IConfiguration configuration)
     {
-        ArgumentNullException.ThrowIfNull(configuration);
-
-        var section = configuration.GetSection(LocalDbRegistration.ReplicationSectionPath);
-
-        return !string.IsNullOrWhiteSpace(section["PeerAddress"])
-            || !string.IsNullOrWhiteSpace(section["ApiKey"]);
+        // Shared with the alarm-historian drain gate (#603): one predicate, so the nodes that capture
+        // changes and the nodes that stand down for a peer can never disagree.
+        return Runtime.LocalDbReplicationPolicy.IsConfigured(configuration);
     }
 
     /// <summary>Opts every replicated table into change capture, baselining what is already there.</summary>

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

@@ -0,0 +1,43 @@
+using Microsoft.Extensions.Configuration;
+
+namespace ZB.MOM.WW.OtOpcUa.Runtime;
+
+/// <summary>
+///     The ONE definition of "this node's LocalDb is replicated to a peer" (#603). Two consumers must agree
+///     on it, and used to carry separate predicates that diverged: the Host's table-registration step
+///     (<c>LocalDbSetup.ReplicationIsConfigured</c> - which nodes capture changes) and the alarm-historian
+///     drain gate (<c>AddAlarmHistorian</c> - which nodes stand down in favour of a peer). Lives in Runtime
+///     because Host references Runtime and not the reverse.
+/// </summary>
+/// <remarks>
+///     <para>
+///         <b>Predicate: <c>LocalDb:Replication:PeerAddress</c> OR <c>LocalDb:Replication:ApiKey</c> is
+///         non-blank.</b> The OR is required: only the dialling half of a pair sets <c>PeerAddress</c>, while
+///         the passive half carries the shared <c>ApiKey</c> (plus a listen port) and nothing else.
+///     </para>
+///     <para>
+///         <b>Why the drain gate takes the same predicate (it previously used PeerAddress OR
+///         <c>SyncListenPort</c>).</b> The gate's question is "does a peer hold the same rows?". The passive
+///         sync endpoint is fail-closed on the key, so a node with a listen port but no <c>ApiKey</c> rejects
+///         every inbound stream and shares its queue with nobody - gating it would park its alarm events
+///         behind a Primary that cannot deliver them. Conversely an <c>ApiKey</c>-only passive node IS
+///         registered and does receive its peer's rows, so it must be gated or both halves drain. The port
+///         is not evidence of a peer; the key is.
+///     </para>
+/// </remarks>
+public static class LocalDbReplicationPolicy
+{
+    /// <summary>Configuration section holding the LocalDb replication options.</summary>
+    public const string ReplicationSectionPath = "LocalDb:Replication";
+
+    /// <summary>Whether this node's LocalDb replicates to a pair peer.</summary>
+    /// <param name="configuration">The application configuration.</param>
+    /// <returns><c>true</c> when <c>PeerAddress</c> or <c>ApiKey</c> is set under the replication section.</returns>
+    public static bool IsConfigured(IConfiguration configuration)
+    {
+        ArgumentNullException.ThrowIfNull(configuration);
+        var section = configuration.GetSection(ReplicationSectionPath);
+        return !string.IsNullOrWhiteSpace(section["PeerAddress"])
+            || !string.IsNullOrWhiteSpace(section["ApiKey"]);
+    }
+}

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

@@ -177,12 +177,11 @@ public static class ServiceCollectionExtensions
         // possibly deliver its events. Scoping the gate to "is my queue actually shared?" is what
         // keeps the two scopes from disagreeing.
         // BOTH halves of a pair share the queue, but only one of them dials: the initiator sets
-        // Replication:PeerAddress, its partner only sets SyncListenPort and waits. Testing the dial
-        // side alone would leave the listening half permanently ungated — one drainer per pair by
-        // accident rather than by role, and the wrong one whenever the roles swap.
-        var replicated =
-            !string.IsNullOrWhiteSpace(configuration["LocalDb:Replication:PeerAddress"])
-            || !string.IsNullOrWhiteSpace(configuration["LocalDb:SyncListenPort"]);
+        // Replication:PeerAddress, its partner carries the shared ApiKey and waits. The predicate is
+        // LocalDbReplicationPolicy - the same one LocalDbSetup uses to decide which nodes capture
+        // changes at all. It deliberately does NOT count SyncListenPort: the passive endpoint is
+        // fail-closed on the key, so a port with no ApiKey shares nothing and must keep draining.
+        var replicated = LocalDbReplicationPolicy.IsConfigured(configuration);
 
         if (!replicated)
         {

tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Historian/AlarmHistorianRegistrationTests.cs

@@ -122,14 +122,18 @@ public sealed class AlarmHistorianRegistrationTests
     /// </remarks>
     [Theory]
     // Neither key: an unpaired node. Its rows exist nowhere else.
-    [InlineData(null, null, true)]
+    [InlineData(null, null, null, true)]
     // The dialing half of a pair — gated, because its partner holds the same rows.
-    [InlineData("http://peer:9001", null, false)]
+    [InlineData("http://peer:9001", null, null, false)]
     // The LISTENING half. Same shared queue, but it never dials, so a PeerAddress-only test would
-    // wrongly class it as unpaired and leave it permanently ungated.
-    [InlineData(null, "9001", false)]
+    // wrongly class it as unpaired and leave it permanently ungated. It is identified by the shared
+    // ApiKey (the same predicate LocalDbSetup registers tables on), not by its listen port.
+    [InlineData(null, "9001", "k", false)]
+    // A listen port with NO key: the fail-closed passive endpoint rejects every inbound stream, so the
+    // queue is shared with nobody and the node must keep draining (#603: this used to be gated).
+    [InlineData(null, "9001", null, true)]
     public async Task Only_an_unreplicated_node_ignores_the_role_view(
-        string? peerAddress, string? syncListenPort, bool expectDrain)
+        string? peerAddress, string? syncListenPort, string? apiKey, bool expectDrain)
     {
         var tempDir = Path.Combine(Path.GetTempPath(), "otopcua-alarmhist-test-" + Guid.NewGuid().ToString("N"));
         Directory.CreateDirectory(tempDir);
@@ -143,6 +147,7 @@ public sealed class AlarmHistorianRegistrationTests
                 ["LocalDb:Path"] = dbPath,
                 ["LocalDb:Replication:PeerAddress"] = peerAddress,
                 ["LocalDb:SyncListenPort"] = syncListenPort,
+                ["LocalDb:Replication:ApiKey"] = apiKey,
             });
 
             var services = BaseServices();

tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Historian/LocalDbReplicationPolicyTests.cs

@@ -0,0 +1,32 @@
+using Microsoft.Extensions.Configuration;
+using Shouldly;
+using Xunit;
+
+namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Historian;
+
+/// <summary>#603: the single "is this node's LocalDb replicated" predicate shared by table registration and the alarm drain gate.</summary>
+public sealed class LocalDbReplicationPolicyTests
+{
+    private static IConfiguration Config(params (string Key, string? Value)[] pairs) =>
+        new ConfigurationBuilder()
+            .AddInMemoryCollection(pairs.Select(p => new KeyValuePair<string, string?>(p.Key, p.Value)))
+            .Build();
+
+    /// <summary>PeerAddress OR ApiKey counts; the listen port alone and blank values do not.</summary>
+    [Theory]
+    [InlineData("LocalDb:Replication:PeerAddress", "http://peer:9001", true)]
+    [InlineData("LocalDb:Replication:ApiKey", "k", true)]
+    [InlineData("LocalDb:SyncListenPort", "9001", false)]
+    [InlineData("LocalDb:Replication:ApiKey", "  ", false)]
+    public void IsConfigured_follows_the_PeerAddress_or_ApiKey_rule(string key, string value, bool expected)
+    {
+        LocalDbReplicationPolicy.IsConfigured(Config((key, value))).ShouldBe(expected);
+    }
+
+    /// <summary>No replication keys at all is an unpaired node.</summary>
+    [Fact]
+    public void IsConfigured_is_false_for_an_empty_configuration()
+    {
+        LocalDbReplicationPolicy.IsConfigured(Config()).ShouldBeFalse();
+    }
+}