Merge #590: ClusterRoleInfo records the RoleLeaderChanged address directly (lagging State snapshot could pin a role leader to null)

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

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<string, Member?> _roleLeaders = new(StringComparer.Ordinal);
+    private readonly Dictionary<string, Akka.Actor.Address?> _roleLeaders = new(StringComparer.Ordinal);
     private readonly Dictionary<string, HashSet<Member>> _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);
         }

src/Core/ZB.MOM.WW.OtOpcUa.Cluster/ZB.MOM.WW.OtOpcUa.Cluster.csproj

@@ -5,6 +5,10 @@
         <TreatWarningsAsErrors>true</TreatWarningsAsErrors>
     </PropertyGroup>
 
+    <ItemGroup>
+        <InternalsVisibleTo Include="ZB.MOM.WW.OtOpcUa.Cluster.Tests"/>
+    </ItemGroup>
+
     <ItemGroup>
         <PackageReference Include="Akka.Hosting"/>
         <PackageReference Include="Akka.Cluster"/>

tests/Core/ZB.MOM.WW.OtOpcUa.Cluster.Tests/ClusterRoleInfoRoleLeaderCaseTests.cs

@@ -126,4 +126,43 @@ public sealed class ClusterRoleInfoRoleLeaderCaseTests
             hostA.Dispose();
         }
     }
+
+    /// <summary>
+    ///     Deterministic pin for Gitea #590: a RoleLeaderChanged event whose leader address is not (yet)
+    ///     in <c>Cluster.State.Members</c> 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.
+    /// </summary>
+    [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<string>(),
+        };
+        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<ClusterRoleInfo>.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();
+        }
+    }
 }