fix(cluster): record RoleLeaderChanged leader from the event, not the lagging state snapshot (#590)
ClusterRoleInfo resolved the leader address through Cluster.State.Members, which is published asynchronously; a miss stored null permanently, so RoleLeader(role) stayed null (flaky test timed out under load: 6/10 failing with CPU contention, 0/20 after). Only the address is ever read, so store it directly. Adds a deterministic unit test (fails without the fix) and InternalsVisibleTo for the Cluster.Tests project. Claude-Session: https://claude.ai/code/session_01Ef2ozxS7QPanGT8uoATbTe
This commit is contained in:
@@ -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);
|
||||
}
|
||||
|
||||
@@ -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"/>
|
||||
|
||||
Reference in New Issue
Block a user