using Akka.Actor; using Akka.Cluster; using Microsoft.Extensions.Logging; using Microsoft.Extensions.Options; using ZB.MOM.WW.OtOpcUa.Commons.Interfaces; using CommonsNodeId = ZB.MOM.WW.OtOpcUa.Commons.Types.NodeId; namespace ZB.MOM.WW.OtOpcUa.Cluster; /// /// Thread-safe live view of cluster membership and role topology. Subscribes to /// , , and /// through an internal subscriber actor and keeps /// a snapshot of role-to-members + role-to-leader. The CLR-facing event surface is /// . /// public sealed class ClusterRoleInfo : IClusterRoleInfo, IDisposable { private readonly Akka.Cluster.Cluster _cluster; private readonly ILogger _logger; private readonly CommonsNodeId _localNode; private readonly HashSet _localRoles; private readonly string? _clusterRole; private readonly string? _clusterId; private readonly object _lock = new(); // R3-04 (01/C-9, census row 17) — the two dictionaries below have DIFFERENT correct comparers, // and that split is deliberate, not an oversight (review finding, 2026-08-20). // // _membersByRole answers "every member carrying (a spelling of) this role" — OrdinalIgnoreCase, // so a PEER running an older build that still gossips a miscased plain role during a rolling // upgrade is not invisible to MembersWithRole("driver"). Merging casings IS the point here. // // _roleLeaders must stay Ordinal. Akka computes — and raises RoleLeaderChanged for — role // leadership PER EXACT GOSSIPED ROLE STRING: "driver" and "Driver" are two entirely distinct // roles to Akka's own role-leader machinery, each independently electing its own leader. An // OrdinalIgnoreCase dictionary here would collapse both onto ONE key, so whichever // 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> _membersByRole = new(StringComparer.OrdinalIgnoreCase); private IActorRef? _subscriber; /// Initializes a new instance of the ClusterRoleInfo class. /// The Akka actor system. /// The cluster configuration options. /// The logger instance. public ClusterRoleInfo(ActorSystem system, IOptions options, ILogger logger) { _cluster = Akka.Cluster.Cluster.Get(system); _logger = logger; // NodeId encodes host:port so cluster members on shared hosts (test loopback, dev VMs // sharing a bind IP) stay distinct. Production hosts have unique DNS names so the port // suffix is harmless redundancy. _localNode = CommonsNodeId.Parse($"{options.Value.PublicHostname}:{options.Value.Port}"); // OrdinalIgnoreCase (Gitea #532): Cluster:Roles binds verbatim, so a node authored with 'Driver' // used to answer HasRole("driver") false while gossiping a role Akka would still match against // its own ordinal comparison — the node was a driver to the cluster and not to itself. _localRoles = new HashSet(options.Value.Roles, StringComparer.OrdinalIgnoreCase); // Derived from the node's OWN configured roles (not live Cluster.State) so this identity is // available at host-build time, before the cluster forms. Iterate the ORIGINAL array — not // _localRoles — because first-wins requires configuration order, which HashSet does not // guarantee. The OTOPCUA_ROLES fallback is already applied to the bound options by // AddOtOpcUaCluster's PostConfigure, so nothing is re-resolved here. var clusterRoles = RoleParser.ClusterRolesIn(options.Value.Roles); if (clusterRoles.Count > 0) { _clusterRole = clusterRoles[0]; _clusterId = RoleParser.ClusterIdFromRole(_clusterRole); if (clusterRoles.Count > 1) { _logger.LogWarning( "Node configured with {Count} cluster-scoped roles ({Roles}); using the first ({ClusterRole})", clusterRoles.Count, string.Join(", ", clusterRoles), _clusterRole); } } SeedFromCurrentState(); _subscriber = system.ActorOf(Props.Create(() => new SubscriberActor(this)), "clusterroleinfo-subscriber"); } /// public CommonsNodeId LocalNode => _localNode; /// public IReadOnlySet LocalRoles => _localRoles; /// public string? ClusterRole => _clusterRole; /// public string? ClusterId => _clusterId; /// public bool HasRole(string role) => _localRoles.Contains(role); /// public IReadOnlyList MembersWithRole(string role) { lock (_lock) { if (!_membersByRole.TryGetValue(role, out var members)) return Array.Empty(); return members .Select(m => ToNodeId(m.Address)) .ToArray(); } } /// public CommonsNodeId? RoleLeader(string role) { lock (_lock) { return _roleLeaders.TryGetValue(role, out var leader) && leader is not null ? ToNodeId(leader) : (CommonsNodeId?)null; } } /// Occurs when the leader for a role changes. public event EventHandler? RoleLeaderChanged; private void SeedFromCurrentState() { var snapshot = _cluster.State; lock (_lock) { foreach (var member in snapshot.Members) foreach (var role in member.Roles) { if (!_membersByRole.TryGetValue(role, out var set)) _membersByRole[role] = set = new HashSet(); set.Add(member); } foreach (var role in snapshot.Members.SelectMany(m => m.Roles).Distinct()) { var leaderAddr = snapshot.RoleLeader(role); _roleLeaders[role] = leaderAddr; } } } /// Handles a cluster member event (member up/removed). /// The member event from the cluster. internal void HandleMemberEvent(ClusterEvent.IMemberEvent evt) { lock (_lock) { switch (evt) { case ClusterEvent.MemberUp up: foreach (var role in up.Member.Roles) { if (!_membersByRole.TryGetValue(role, out var set)) _membersByRole[role] = set = new HashSet(); set.Add(up.Member); } break; // Cluster-002: a leaving / exiting / downed member is no longer a usable member, so // drop it as soon as that transition lands rather than waiting for MemberRemoved // (several seconds later). MembersWithRole now has a production caller (the // historization scope-peer check), which must not defer to a departing partner. // Member equality is by UniqueAddress, so the status change does not matter here. case ClusterEvent.MemberLeft left: RemoveMember(left.Member); break; case ClusterEvent.MemberExited exited: RemoveMember(exited.Member); break; case ClusterEvent.MemberDowned downed: RemoveMember(downed.Member); break; case ClusterEvent.MemberRemoved removed: RemoveMember(removed.Member); break; } } } // Caller holds _lock. private void RemoveMember(Member member) { foreach (var role in member.Roles) if (_membersByRole.TryGetValue(role, out var set)) set.Remove(member); } /// Handles a role leader change event. /// The role leader changed event from the cluster. internal void HandleRoleLeaderChanged(ClusterEvent.RoleLeaderChanged evt) { CommonsNodeId? previous = null; CommonsNodeId? next = null; var raise = false; lock (_lock) { _roleLeaders.TryGetValue(evt.Role, out var prevLeader); if (prevLeader is not null) previous = ToNodeId(prevLeader); // 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); } if (!raise) return; try { RoleLeaderChanged?.Invoke(this, new RoleLeaderChangedEventArgs { Role = evt.Role, PreviousLeader = previous, NewLeader = next, }); } catch (Exception ex) { _logger.LogWarning(ex, "RoleLeaderChanged subscriber threw for role {Role}", evt.Role); } } private static CommonsNodeId ToNodeId(Akka.Actor.Address address) => CommonsNodeId.Parse($"{address.Host ?? string.Empty}:{address.Port ?? 0}"); /// Disposes the ClusterRoleInfo and stops the subscriber actor. public void Dispose() { _subscriber?.Tell(PoisonPill.Instance); _subscriber = null; } private sealed class SubscriberActor : ReceiveActor { /// Initializes a new instance of the SubscriberActor class. /// The ClusterRoleInfo instance to forward events to. public SubscriberActor(ClusterRoleInfo owner) { Receive(e => owner.HandleMemberEvent(e)); Receive(e => owner.HandleRoleLeaderChanged(e)); // LeaderChanged is intentionally a no-op here: ServiceLevel lives in RedundancyStateActor // (per-cluster mesh Phase 6: a cluster-{ClusterId}-scoped singleton spawned on every // driver node, no longer the admin-role singleton it once was) and has no consumer on this // ClusterRoleInfo path by design. Receive(_ => { }); Receive(_ => { /* seeded from initial snapshot */ }); } /// protected override void PreStart() { Akka.Cluster.Cluster.Get(Context.System).Subscribe( Self, ClusterEvent.InitialStateAsEvents, typeof(ClusterEvent.IMemberEvent), typeof(ClusterEvent.LeaderChanged), typeof(ClusterEvent.RoleLeaderChanged)); } /// protected override void PostStop() => Akka.Cluster.Cluster.Get(Context.System).Unsubscribe(Self); } }