Skip to content

Fix DistributeByGroupAffinity assigning a bumped projection version to nodes that cannot build it #3792

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
jeremydmiller merged 5 commits into from
Aug 3, 2026
+469 −76
Merged

Fix DistributeByGroupAffinity assigning a bumped projection version to nodes that cannot build it #3792

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
1beed4e
Split a group by capability set so a version bump lands on its own nodes
erdtsieck Aug 3, 2026
ba0ebc1
Keep a split group on as few nodes as its capability split allows
erdtsieck Aug 3, 2026
f2e0f6f
Add an end-to-end test over a real multi-database Marten store
erdtsieck Aug 3, 2026
bd513eb
camelCase the two new private helpers
erdtsieck Aug 3, 2026
2ea368f
Model the sharded, tenant-partitioned shape in the end-to-end test
erdtsieck Aug 3, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
80 changes: 80 additions & 0 deletions src/Testing/CoreTests/Runtime/Agents/distribute_by_group_affinity.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -235,5 +235,85 @@ public void an_incumbent_node_keeps_its_group_up_to_the_ceiling()
grid.AgentFor(g2).AssignedNode.ShouldBe(node2, "an under-ceiling incumbent keeps its group");
grid.AgentFor(g3).AssignedNode.ShouldBe(node3, "only the over-ceiling group moves, to the empty node");
}

// event-subscriptions://{type}/{name}/{databaseId}/{projection}/{shardKey}/v{version}/{tenant}
// — the real EventSubscriptionAgentFamily.UriFor grammar, so the version sits in its own segment.
private static Uri VersionedAgent(string db, uint version, string tenant) =>
new($"event-subscriptions://marten/main/{db}/Proj/All/v{version}/{tenant}");

[Fact]
public void a_version_bump_splits_a_group_between_the_old_and_new_version_nodes()
{
// A projection version bump on a sharded store: one shard database's group spans the previous
// version's agent — declared by, and RUNNING on, the blue node — and the new version's agent,
// declared only by the green node. No node is capable of the whole group, so the members must
// fall back individually: the old version stays on blue, the new version goes to green.
//
// This is the intersection of the two cases above, and it is what a blue/green deployment of a
// sharded store looks like at every evaluation for the whole rollout, not just transiently.
var previous = VersionedAgent("db1", 22, "t1");
var bumped = VersionedAgent("db1", 23, "t1");

var grid = new AssignmentGrid();
var blue = grid.WithNode(1, Guid.NewGuid()).HasCapabilities(new[] { previous });
blue.Running(previous);
var green = grid.WithNode(2, Guid.NewGuid()).HasCapabilities(new[] { bumped });

grid.WithAgents(previous, bumped);

grid.DistributeByGroupAffinity("event-subscriptions", DatabaseKey);

grid.AgentFor(previous).AssignedNode.ShouldBe(blue,
"the previous version keeps running where it is");
grid.AgentFor(bumped).AssignedNode.ShouldBe(green,
"the new version's agent may only run on the node that declares it — the blue node cannot build it");
}

[Fact]
public void a_split_group_costs_only_as_many_nodes_as_the_capability_split_forces()
{
// The same version bump, now with the projections whose version did NOT change also in the group.
// Every node declares those, so nothing stops them landing on a third node — and what a shard
// database costs in connection pools is the number of DISTINCT nodes holding any of its agents, so
// a third host is a third pool set on that database. A split group must still occupy only as many
// nodes as the capability split forces: one per version.
var databases = new[] { "db1", "db2", "db3", "db4" };
var tenants = new[] { "t1", "t2" };

Uri Unchanged(string db, string tenant) =>
new($"event-subscriptions://marten/main/{db}/Other/All/v7/{tenant}");

Uri[] Across(Func<string, string, Uri> agent) =>
databases.SelectMany(db => tenants.Select(t => agent(db, t))).ToArray();

var previous = Across((db, t) => VersionedAgent(db, 22, t));
var bumped = Across((db, t) => VersionedAgent(db, 23, t));
var unchanged = Across(Unchanged);

var grid = new AssignmentGrid();
var blues = new[] { grid.WithNode(1, Guid.NewGuid()), grid.WithNode(2, Guid.NewGuid()) };
var greens = new[] { grid.WithNode(3, Guid.NewGuid()), grid.WithNode(4, Guid.NewGuid()) };

foreach (var blue in blues) blue.HasCapabilities(previous.Concat(unchanged));
foreach (var green in greens) green.HasCapabilities(bumped.Concat(unchanged));

grid.WithAgents(previous.Concat(bumped).Concat(unchanged).ToArray());

grid.DistributeByGroupAffinity("event-subscriptions", DatabaseKey);

foreach (var db in databases)
{
var hosts = tenants
.SelectMany(t => new[] { VersionedAgent(db, 22, t), VersionedAgent(db, 23, t), Unchanged(db, t) })
.Select(uri => grid.AgentFor(uri).AssignedNode)
.Distinct()
.ToList();

hosts.Count.ShouldBe(2,
$"{db} must sit on exactly two nodes — one per version — so it attracts two pool sets, not three");
hosts.ShouldContain(host => blues.Contains(host!), $"{db}'s previous version must be on a blue node");
hosts.ShouldContain(host => greens.Contains(host!), $"{db}'s new version must be on a green node");
}
}
}

207 changes: 131 additions & 76 deletions src/Wolverine/Runtime/Agents/AssignmentGrid.Distribution.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -178,103 +178,158 @@ public void DistributeByGroupAffinity(string scheme, Func<Uri, string> groupKey,

foreach (var group in groups)
{
var members = group.ToList();

// Candidate nodes for the whole group: nodes capable of running every member (all nodes when
// capabilities are homogeneous) — plus any node that was already running part of the group
// when the grid was assembled. The grandfathering mirrors the even paths, which leave running
// agents in place regardless of declared capabilities: a node's capability snapshot is
// persisted once at node startup, so a node that started before (say) a tenant database was
// provisioned never declares that database's agents even though it is happily running them.
var candidates = sameCapabilities
? nodes
: nodes.Where(n => members.All(m => m.CandidateNodes.Contains(n))
|| members.Any(m => m.OriginalNode == n)).ToList();

if (candidates.Count == 0)
// A group is one placement unit — except under mixed capabilities, where members declared by
// different sets of nodes cannot share a host at all. That is what a blue/green rollout of a
// multi-database store looks like: one shard database's group spans the previous version's
// agents (only the blue nodes can build them) and the new version's (only the green nodes can),
// so no node is capable of the whole group and the group has to split. Partitioning by
// capability set keeps affinity inside a version — a database still has one owner per version,
// not one per agent. With homogeneous capabilities there is exactly one partition, so the
// common path is unchanged.
var partitions = sameCapabilities
? [group.ToList()]
: group
.GroupBy(CapabilityKey)
.OrderByDescending(partition => partition.Count())
Comment thread
erdtsieck marked this conversation as resolved.
.ThenBy(partition => partition.Key, StringComparer.Ordinal)
.Select(partition => partition.ToList())
.ToList();

// Partitions of one group prefer to land on the same node as their siblings: what a database
// costs in connection pools is the number of DISTINCT nodes hosting any of its agents, so a
// split group should still occupy as few nodes as its capability split forces — two during a
// version bump (one per version), not one per partition.
var siblingHosts = new List<Node>();

foreach (var members in partitions)
{
// GH-3341: a whole group whose members are all unassigned AND declared by no node is a
// stale-snapshot artifact, not a genuine blue/green gap. A node captures its
// event-subscription capabilities once at startup (StartLocalAgentProcessingAsync), so a
// shard database provisioned after every surviving node started is absent from all their
// snapshots even though every node can run it — the agents are still enumerated as
// supported by AllKnownAgentsAsync. When such a group's incumbent was a departed node, the
// OriginalNode grandfathering above cannot rescue it, and the per-member fallback below
// would park every member: the shard silently stops projecting with no running agent, no
// log, and no self-heal until a restart refreshes the snapshots. Treat the whole group as
// assignable to any node so it always has a home, kept together to preserve the
// connection-pool affinity this method exists to provide.
if (members.All(m => m.AssignedNode == null && m.CandidateNodes.Count == 0))
// Candidate nodes for the whole partition: nodes capable of running every member (all nodes
// when capabilities are homogeneous) — plus any node that was already running part of it
// when the grid was assembled. The grandfathering mirrors the even paths, which leave
// running agents in place regardless of declared capabilities: a node's capability snapshot
// is persisted once at node startup, so a node that started before (say) a tenant database
// was provisioned never declares that database's agents even though it is happily running
// them.
var candidates = sameCapabilities
? nodes
: nodes.Where(n => members.All(m => m.CandidateNodes.Contains(n))
|| members.Any(m => m.OriginalNode == n)).ToList();

if (candidates.Count == 0)
{
candidates = nodes;
}
else
{
// Mixed capabilities (genuine blue/green): an already-running member stays where it is
// (minimal disruption), an unassigned member with a capable node goes to its
// least-loaded one, and an unassigned member no node declares falls back to the
// least-loaded node overall rather than being silently stranded (GH-3341).
foreach (var member in members)
// GH-3341: a whole group whose members are all unassigned AND declared by no node is a
// stale-snapshot artifact, not a genuine blue/green gap. A node captures its
// event-subscription capabilities once at startup (StartLocalAgentProcessingAsync), so a
// shard database provisioned after every surviving node started is absent from all their
// snapshots even though every node can run it — the agents are still enumerated as
// supported by AllKnownAgentsAsync. When such a group's incumbent was a departed node,
// the OriginalNode grandfathering above cannot rescue it, and the per-member fallback
// below would park every member: the shard silently stops projecting with no running
// agent, no log, and no self-heal until a restart refreshes the snapshots. Treat the
// whole group as assignable to any node so it always has a home, kept together to
// preserve the connection-pool affinity this method exists to provide.
if (members.All(m => m.AssignedNode == null && m.CandidateNodes.Count == 0))
{
candidates = nodes;
}
else
{
if (member.AssignedNode != null)
// An already-running member stays where it is (minimal disruption), an unassigned
// member with a capable node goes to its least-loaded one, and an unassigned member
// no node declares falls back to the least-loaded node overall rather than being
// silently stranded (GH-3341).
foreach (var member in members)
{
load[member.AssignedNode] = load.GetValueOrDefault(member.AssignedNode) + 1;
continue;
}

var candidate = member.CandidateNodes
.OrderBy(n => load.GetValueOrDefault(n))
.ThenBy(n => n.IsLeader)
.ThenBy(n => n.AssignedId)
.FirstOrDefault()
?? nodes
if (member.AssignedNode != null)
{
load[member.AssignedNode] = load.GetValueOrDefault(member.AssignedNode) + 1;
Remember(siblingHosts, member.AssignedNode);
continue;
}

var candidate = member.CandidateNodes
.OrderBy(n => load.GetValueOrDefault(n))
.ThenBy(n => n.IsLeader)
.ThenBy(n => n.AssignedId)
.First();
.FirstOrDefault()
?? nodes
.OrderBy(n => load.GetValueOrDefault(n))
.ThenBy(n => n.IsLeader)
.ThenBy(n => n.AssignedId)
.First();

candidate.Assign(member);
load[candidate] += 1;
Remember(siblingHosts, candidate);
}

candidate.Assign(member);
load[candidate] += 1;
continue;
}
}

// Minimal disruption, mirroring DistributeEvenly: the node already running the WHOLE
// partition keeps it as long as that doesn't push the node past the ceiling. Without this,
// every evaluation reshuffles groups from scratch and a node whose stale capability snapshot
// keeps it out of the capability candidates can be starved permanently across evaluations.
var incumbent = members[0].AssignedNode;
if (incumbent != null && members.Any(m => m.AssignedNode != incumbent))
{
incumbent = null;
}

if (incumbent != null && candidates.Contains(incumbent) &&
load[incumbent] + members.Count <= maximum)
{
load[incumbent] += members.Count;
Remember(siblingHosts, incumbent);
continue;
}
}

// Minimal disruption, mirroring DistributeEvenly: the node already running the WHOLE group
// keeps it as long as that doesn't push the node past the ceiling. Without this, every
// evaluation reshuffles groups from scratch and a node whose stale capability snapshot keeps
// it out of the capability candidates can be starved permanently across evaluations.
var incumbent = members[0].AssignedNode;
if (incumbent != null && members.Any(m => m.AssignedNode != incumbent))
{
incumbent = null;
}
// Otherwise the least-loaded candidate hosts the whole partition — preferring a node that
// already hosts a sibling partition of this same group, so a split group still costs as few
// connection pools per database as its capability split allows (tie-breaks: non-leader
// first, then node id).
var node = candidates
.Where(n => siblingHosts.Contains(n) && load[n] + members.Count <= maximum)
.OrderBy(n => load[n])
.ThenBy(n => n.IsLeader)
.ThenBy(n => n.AssignedId)
.FirstOrDefault()
?? candidates
.OrderBy(n => load[n])
.ThenBy(n => n.IsLeader)
.ThenBy(n => n.AssignedId)
.First();

foreach (var agent in members)
{
node.Assign(agent);
}

if (incumbent != null && candidates.Contains(incumbent) &&
load[incumbent] + members.Count <= maximum)
{
load[incumbent] += members.Count;
continue;
load[node] += members.Count;
Remember(siblingHosts, node);
}
}

// Otherwise the least-loaded candidate hosts the whole group (tie-breaks: non-leader first,
// then node id).
var node = candidates
.OrderBy(n => load[n])
.ThenBy(n => n.IsLeader)
.ThenBy(n => n.AssignedId)
.First();

foreach (var agent in members)
static void Remember(List<Node> hosts, Node node)
{
if (!hosts.Contains(node))
{
node.Assign(agent);
hosts.Add(node);
}

load[node] += members.Count;
}
}

/// <summary>
/// Stable identity of the set of nodes that declare an agent as a capability, used to sub-partition a
/// group in <see cref="DistributeByGroupAffinity(string, Func{Uri, string}, Func{Uri, bool})" />. Agents
/// with the same key can share a host; agents with different keys generally cannot, which is exactly the
/// blue/green split. An agent no node declares gets the empty key, so those stay together and keep the
/// GH-3341 whole-group rescue.
/// </summary>
private static string CapabilityKey(Agent agent) =>
string.Join(",", agent.CandidateNodes.Select(n => n.AssignedId).OrderBy(id => id));

public bool AllNodesHaveSameCapabilities(string scheme)
{
return AllNodesHaveSameCapabilities(scheme, _ => true);
Expand Down
Loading