diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Cluster/ClusterRoleInfo.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Cluster/ClusterRoleInfo.cs index 689e435b..7dfd2fdd 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Cluster/ClusterRoleInfo.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Cluster/ClusterRoleInfo.cs @@ -37,7 +37,7 @@ public sealed class ClusterRoleInfo : IClusterRoleInfo, IDisposable // RoleLeaderChanged event landed last would silently overwrite the other's leader — turning two // real, independent facts Akka tracks separately into one, wrong, merged one. Pinned by // ClusterRoleInfoRoleLeaderCaseTests.RoleLeaderChanged_for_driver_then_Driver_keeps_distinct_entries. - private readonly Dictionary _roleLeaders = new(StringComparer.Ordinal); + private readonly Dictionary _roleLeaders = new(StringComparer.Ordinal); private readonly Dictionary> _membersByRole = new(StringComparer.OrdinalIgnoreCase); private IActorRef? _subscriber; @@ -113,7 +113,7 @@ public sealed class ClusterRoleInfo : IClusterRoleInfo, IDisposable lock (_lock) { return _roleLeaders.TryGetValue(role, out var leader) && leader is not null - ? ToNodeId(leader.Address) + ? ToNodeId(leader) : (CommonsNodeId?)null; } } @@ -137,9 +137,7 @@ public sealed class ClusterRoleInfo : IClusterRoleInfo, IDisposable foreach (var role in snapshot.Members.SelectMany(m => m.Roles).Distinct()) { var leaderAddr = snapshot.RoleLeader(role); - _roleLeaders[role] = leaderAddr is not null - ? snapshot.Members.FirstOrDefault(m => m.Address == leaderAddr) - : null; + _roleLeaders[role] = leaderAddr; } } } @@ -179,17 +177,17 @@ public sealed class ClusterRoleInfo : IClusterRoleInfo, IDisposable lock (_lock) { - _roleLeaders.TryGetValue(evt.Role, out var prevMember); - if (prevMember is not null) - previous = ToNodeId(prevMember.Address); + _roleLeaders.TryGetValue(evt.Role, out var prevLeader); + if (prevLeader is not null) + previous = ToNodeId(prevLeader); - var nextMember = evt.Leader is null - ? null - : _cluster.State.Members.FirstOrDefault(m => m.Address == evt.Leader); - - _roleLeaders[evt.Role] = nextMember; - if (nextMember is not null) - next = ToNodeId(nextMember.Address); + // Record the event's own address. The leader used to be resolved by looking the address up + // in Cluster.State.Members, but that snapshot is published asynchronously and can lag the + // event: a miss stored null and nothing re-ran, so RoleLeader(role) stayed null for good + // (Gitea #590). Only the address is ever read, so no Member lookup is needed. + _roleLeaders[evt.Role] = evt.Leader; + if (evt.Leader is not null) + next = ToNodeId(evt.Leader); raise = !Nullable.Equals(previous, next); } diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Cluster/ZB.MOM.WW.OtOpcUa.Cluster.csproj b/src/Core/ZB.MOM.WW.OtOpcUa.Cluster/ZB.MOM.WW.OtOpcUa.Cluster.csproj index 353c9628..2d45cdf5 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Cluster/ZB.MOM.WW.OtOpcUa.Cluster.csproj +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Cluster/ZB.MOM.WW.OtOpcUa.Cluster.csproj @@ -5,6 +5,10 @@ true + + + + diff --git a/tests/Core/ZB.MOM.WW.OtOpcUa.Cluster.Tests/ClusterRoleInfoRoleLeaderCaseTests.cs b/tests/Core/ZB.MOM.WW.OtOpcUa.Cluster.Tests/ClusterRoleInfoRoleLeaderCaseTests.cs index 82c3aee0..628f756b 100644 --- a/tests/Core/ZB.MOM.WW.OtOpcUa.Cluster.Tests/ClusterRoleInfoRoleLeaderCaseTests.cs +++ b/tests/Core/ZB.MOM.WW.OtOpcUa.Cluster.Tests/ClusterRoleInfoRoleLeaderCaseTests.cs @@ -126,4 +126,43 @@ public sealed class ClusterRoleInfoRoleLeaderCaseTests hostA.Dispose(); } } + + /// + /// Deterministic pin for Gitea #590: a RoleLeaderChanged event whose leader address is not (yet) + /// in Cluster.State.Members must still record that leader. The handler used to resolve the + /// address through the state snapshot, which is published asynchronously and can lag the event; + /// a miss stored null permanently. Here the cluster never joins anything, so the leader address + /// is guaranteed absent from the snapshot — the lagging-snapshot case, made exact. + /// + [Fact] + public async Task RoleLeaderChanged_is_recorded_even_when_the_leader_is_not_in_the_state_snapshot() + { + var port = FreePort(); + var options = new AkkaClusterOptions + { + SystemName = SystemName, Hostname = "127.0.0.1", PublicHostname = "127.0.0.1", + Port = port, Roles = new[] { "driver" }, SeedNodes = Array.Empty(), + }; + var system = Akka.Actor.ActorSystem.Create(SystemName, Akka.Configuration.ConfigurationFactory.ParseString( + "akka.actor.provider = cluster\nakka.remote.dot-netty.tcp.hostname = 127.0.0.1\n" + + $"akka.remote.dot-netty.tcp.port = {port}")); + try + { + using var info = new ClusterRoleInfo( + system, Options.Create(options), + Microsoft.Extensions.Logging.Abstractions.NullLogger.Instance); + var driver = new Akka.Actor.Address("akka.tcp", SystemName, "127.0.0.1", 11111); + var capital = new Akka.Actor.Address("akka.tcp", SystemName, "127.0.0.1", 22222); + + info.HandleRoleLeaderChanged(new Akka.Cluster.ClusterEvent.RoleLeaderChanged("driver", driver)); + info.HandleRoleLeaderChanged(new Akka.Cluster.ClusterEvent.RoleLeaderChanged("Driver", capital)); + + info.RoleLeader("driver")!.Value.ToString().ShouldContain("11111"); + info.RoleLeader("Driver")!.Value.ToString().ShouldContain("22222"); + } + finally + { + await system.Terminate(); + } + } }