From 4f6231fa403e16d7278ffdb6f47c32ac8bbc86ba Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 8 Oct 2026 06:56:00 -0400 Subject: [PATCH] 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 --- .../Configuration/LocalDbRegistration.cs | 2 +- .../Configuration/LocalDbSetup.cs | 9 ++-- .../LocalDbReplicationPolicy.cs | 43 +++++++++++++++++++ .../ServiceCollectionExtensions.cs | 11 +++-- .../AlarmHistorianRegistrationTests.cs | 15 ++++--- .../LocalDbReplicationPolicyTests.cs | 32 ++++++++++++++ 6 files changed, 94 insertions(+), 18 deletions(-) create mode 100644 src/Server/ZB.MOM.WW.OtOpcUa.Runtime/LocalDbReplicationPolicy.cs create mode 100644 tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Historian/LocalDbReplicationPolicyTests.cs diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/LocalDbRegistration.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/LocalDbRegistration.cs index f8a54971..a26ac5bb 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/LocalDbRegistration.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/LocalDbRegistration.cs @@ -40,7 +40,7 @@ public static class LocalDbRegistration public const string LocalDbSectionPath = "LocalDb"; /// Configuration section holding ReplicationOptions. - public const string ReplicationSectionPath = "LocalDb:Replication"; + public const string ReplicationSectionPath = Runtime.LocalDbReplicationPolicy.ReplicationSectionPath; /// /// Configuration key carrying the dedicated h2c sync listener's port. Zero or absent means diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/LocalDbSetup.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/LocalDbSetup.cs index 11b1c8d9..7bc8954a 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/LocalDbSetup.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/LocalDbSetup.cs @@ -170,12 +170,9 @@ public static class LocalDbSetup /// true when this node is configured to replicate. 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); } /// Opts every replicated table into change capture, baselining what is already there. diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/LocalDbReplicationPolicy.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/LocalDbReplicationPolicy.cs new file mode 100644 index 00000000..e4106f55 --- /dev/null +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/LocalDbReplicationPolicy.cs @@ -0,0 +1,43 @@ +using Microsoft.Extensions.Configuration; + +namespace ZB.MOM.WW.OtOpcUa.Runtime; + +/// +/// 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 +/// (LocalDbSetup.ReplicationIsConfigured - which nodes capture changes) and the alarm-historian +/// drain gate (AddAlarmHistorian - which nodes stand down in favour of a peer). Lives in Runtime +/// because Host references Runtime and not the reverse. +/// +/// +/// +/// Predicate: LocalDb:Replication:PeerAddress OR LocalDb:Replication:ApiKey is +/// non-blank. The OR is required: only the dialling half of a pair sets PeerAddress, while +/// the passive half carries the shared ApiKey (plus a listen port) and nothing else. +/// +/// +/// Why the drain gate takes the same predicate (it previously used PeerAddress OR +/// SyncListenPort). 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 ApiKey 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 ApiKey-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. +/// +/// +public static class LocalDbReplicationPolicy +{ + /// Configuration section holding the LocalDb replication options. + public const string ReplicationSectionPath = "LocalDb:Replication"; + + /// Whether this node's LocalDb replicates to a pair peer. + /// The application configuration. + /// true when PeerAddress or ApiKey is set under the replication section. + public static bool IsConfigured(IConfiguration configuration) + { + ArgumentNullException.ThrowIfNull(configuration); + var section = configuration.GetSection(ReplicationSectionPath); + return !string.IsNullOrWhiteSpace(section["PeerAddress"]) + || !string.IsNullOrWhiteSpace(section["ApiKey"]); + } +} diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ServiceCollectionExtensions.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ServiceCollectionExtensions.cs index ee611e63..a530ce62 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ServiceCollectionExtensions.cs +++ b/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) { diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Historian/AlarmHistorianRegistrationTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Historian/AlarmHistorianRegistrationTests.cs index 32531012..91f655da 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Historian/AlarmHistorianRegistrationTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Historian/AlarmHistorianRegistrationTests.cs @@ -122,14 +122,18 @@ public sealed class AlarmHistorianRegistrationTests /// [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(); diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Historian/LocalDbReplicationPolicyTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Historian/LocalDbReplicationPolicyTests.cs new file mode 100644 index 00000000..0ff37296 --- /dev/null +++ b/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; + +/// #603: the single "is this node's LocalDb replicated" predicate shared by table registration and the alarm drain gate. +public sealed class LocalDbReplicationPolicyTests +{ + private static IConfiguration Config(params (string Key, string? Value)[] pairs) => + new ConfigurationBuilder() + .AddInMemoryCollection(pairs.Select(p => new KeyValuePair(p.Key, p.Value))) + .Build(); + + /// PeerAddress OR ApiKey counts; the listen port alone and blank values do not. + [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); + } + + /// No replication keys at all is an unpaired node. + [Fact] + public void IsConfigured_is_false_for_an_empty_configuration() + { + LocalDbReplicationPolicy.IsConfigured(Config()).ShouldBeFalse(); + } +}