From 1beed4eb6ec21353a4973f4833b13329ce3c4df2 Mon Sep 17 00:00:00 2001 From: Anne Erdtsieck Date: Mon, 3 Aug 2026 10:45:31 +0200 Subject: [PATCH 1/5] Split a group by capability set so a version bump lands on its own nodes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit DistributeByGroupAffinity claims to mirror DistributeEvenlyWithBlueGreenSemantics, but on a real blue/green rollout of a multi-database store it does the opposite. One shard database's group spans the previous version's agents — declared by, and running on, the blue nodes — and the new version's agents, declared only by the green nodes. No node is capable of the whole group, so the candidate set should be empty and the per-member fallback should place each agent on a capable node. It isn't empty: the OriginalNode grandfathering added for stale capability snapshots keeps the incumbent blue node as a candidate, the fallback is skipped, and the whole group — new version included — is assigned to a node whose store does not register that version. BuildAgentAsync then throws for every one of those agents, the new version never starts anywhere, and the rollout silently does not happen. The grandfathering itself is right; what was missing is that a group is only a single placement unit when its members can actually share a host. Members are now sub-partitioned by the set of nodes that declare them, and each partition placed whole. That keeps affinity inside a version — a database still has one owner per version, which is what the connection-pool budget depends on — while letting the two versions land on their own nodes. With homogeneous capabilities there is one partition per group, so the common path is untouched. Covered by a test that is the intersection of the two existing capability tests: no node can host the whole group AND an incumbent is running part of it. It fails on main with the new version's agent assigned to the blue node. --- .../Agents/distribute_by_group_affinity.cs | 33 +++++++++++++++++++ .../Agents/AssignmentGrid.Distribution.cs | 29 ++++++++++++++-- 2 files changed, 59 insertions(+), 3 deletions(-) diff --git a/src/Testing/CoreTests/Runtime/Agents/distribute_by_group_affinity.cs b/src/Testing/CoreTests/Runtime/Agents/distribute_by_group_affinity.cs index a41d1cdf0..6f83ac2e0 100644 --- a/src/Testing/CoreTests/Runtime/Agents/distribute_by_group_affinity.cs +++ b/src/Testing/CoreTests/Runtime/Agents/distribute_by_group_affinity.cs @@ -235,5 +235,38 @@ 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"); + } } diff --git a/src/Wolverine/Runtime/Agents/AssignmentGrid.Distribution.cs b/src/Wolverine/Runtime/Agents/AssignmentGrid.Distribution.cs index c27f35e90..f940e3d57 100644 --- a/src/Wolverine/Runtime/Agents/AssignmentGrid.Distribution.cs +++ b/src/Wolverine/Runtime/Agents/AssignmentGrid.Distribution.cs @@ -170,15 +170,28 @@ public void DistributeByGroupAffinity(string scheme, Func groupKey, // node whole — groups are indivisible by design. var maximum = (int)Math.Ceiling((double)agents.Count / nodes.Count); + // A group is one placement unit — except under mixed capabilities, where members that are declared + // by disjoint 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. Sub-partitioning by capability set keeps affinity inside a version + // — a database still has one owner per version, not one per agent — while letting the versions land + // on their own nodes. With homogeneous capabilities there is exactly one partition per group, so + // this is a no-op on the common path. var groups = agents .GroupBy(a => groupKey(a.Uri)) - .OrderByDescending(g => g.Count()) - .ThenBy(g => g.Key, StringComparer.Ordinal) + .SelectMany(group => sameCapabilities + ? [(Key: group.Key, Members: group.ToList())] + : group + .GroupBy(CapabilityKey) + .Select(partition => (Key: $"{group.Key}|{partition.Key}", Members: partition.ToList()))) + .OrderByDescending(unit => unit.Members.Count) + .ThenBy(unit => unit.Key, StringComparer.Ordinal) .ToList(); foreach (var group in groups) { - var members = group.ToList(); + var members = group.Members; // 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 @@ -275,6 +288,16 @@ public void DistributeByGroupAffinity(string scheme, Func groupKey, } } + /// + /// Stable identity of the set of nodes that declare an agent as a capability, used to sub-partition a + /// group in . 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. + /// + 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); From ba0ebc1e2b784f2ec09271a2ff014c3c8587fdfa Mon Sep 17 00:00:00 2001 From: Anne Erdtsieck Date: Mon, 3 Aug 2026 11:53:03 +0200 Subject: [PATCH 2/5] Keep a split group on as few nodes as its capability split allows MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Partitioning by capability set fixes the assignment, but it can make the thing this method exists for worse. A shard database's group during a version bump has three partitions, not two: the previous version (blue-only), the new version (green-only), and every projection whose version did not change — and that last one is declared by every node, so it is free to land on a third node and put a third connection pool set on that database. What a database costs is the number of distinct nodes holding any of its agents. So partitions of one group now prefer a node that already hosts a sibling partition, bounded by the same per-node ceiling the incumbent rule already respects. A split group settles on one node per version, and the unchanged projections ride along with one of them instead of claiming a node of their own. The new test asserts the host count per database rather than a placement, and fails without the preference: four databases across two blue and two green nodes spread onto three nodes each. --- .../Agents/distribute_by_group_affinity.cs | 47 ++++ .../Agents/AssignmentGrid.Distribution.cs | 214 ++++++++++-------- 2 files changed, 170 insertions(+), 91 deletions(-) diff --git a/src/Testing/CoreTests/Runtime/Agents/distribute_by_group_affinity.cs b/src/Testing/CoreTests/Runtime/Agents/distribute_by_group_affinity.cs index 6f83ac2e0..f099b178a 100644 --- a/src/Testing/CoreTests/Runtime/Agents/distribute_by_group_affinity.cs +++ b/src/Testing/CoreTests/Runtime/Agents/distribute_by_group_affinity.cs @@ -268,5 +268,52 @@ public void a_version_bump_splits_a_group_between_the_old_and_new_version_nodes( 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 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"); + } + } } diff --git a/src/Wolverine/Runtime/Agents/AssignmentGrid.Distribution.cs b/src/Wolverine/Runtime/Agents/AssignmentGrid.Distribution.cs index f940e3d57..aea0a7a08 100644 --- a/src/Wolverine/Runtime/Agents/AssignmentGrid.Distribution.cs +++ b/src/Wolverine/Runtime/Agents/AssignmentGrid.Distribution.cs @@ -170,121 +170,153 @@ public void DistributeByGroupAffinity(string scheme, Func groupKey, // node whole — groups are indivisible by design. var maximum = (int)Math.Ceiling((double)agents.Count / nodes.Count); - // A group is one placement unit — except under mixed capabilities, where members that are declared - // by disjoint 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. Sub-partitioning by capability set keeps affinity inside a version - // — a database still has one owner per version, not one per agent — while letting the versions land - // on their own nodes. With homogeneous capabilities there is exactly one partition per group, so - // this is a no-op on the common path. var groups = agents .GroupBy(a => groupKey(a.Uri)) - .SelectMany(group => sameCapabilities - ? [(Key: group.Key, Members: group.ToList())] - : group - .GroupBy(CapabilityKey) - .Select(partition => (Key: $"{group.Key}|{partition.Key}", Members: partition.ToList()))) - .OrderByDescending(unit => unit.Members.Count) - .ThenBy(unit => unit.Key, StringComparer.Ordinal) + .OrderByDescending(g => g.Count()) + .ThenBy(g => g.Key, StringComparer.Ordinal) .ToList(); foreach (var group in groups) { - var members = group.Members; - - // 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()) + .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(); + + 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 hosts, Node node) + { + if (!hosts.Contains(node)) { - node.Assign(agent); + hosts.Add(node); } - - load[node] += members.Count; } } From f2e0f6f4941a1af861e55aaabe386dd92b95cae2 Mon Sep 17 00:00:00 2001 From: Anne Erdtsieck Date: Mon, 3 Aug 2026 12:39:20 +0200 Subject: [PATCH 3/5] Add an end-to-end test over a real multi-database Marten store MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The two unit tests drive AssignmentGrid.DistributeByGroupAffinity directly, so they prove the placement rule but not that a real store reaches it. This goes through EventSubscriptionAgentFamily.EvaluateAssignmentsAsync instead, which adds the two steps that sit between a leader and the placement: the per-store pass selection keyed off IEventStore.DatabaseCardinality, and RetireSupersededAgents. The scenario is the rollout itself, over three real tenant databases. The blue store registers a projection at V2 and is already running its agents; the green store registers the same projection at V3 and one other projection unchanged, so the group for each database spans a blue-only, a green-only and an every-node partition exactly as it does in production. Capabilities come from the family's own SupportedAgentsAsync, the grid is seeded from their union the way NodeAgentController does, and the family that evaluates is the BLUE one — so the green agents are reachable only through persisted node capabilities, which is the case that broke. It fails on main with the new version's agents on a blue node: event-subscriptions://marten/main/localhost.bgtenant3/trip/all/v3 may only run on a node that declares it The pool bound stays in the unit test rather than here: it needs agents to outnumber nodes the way a real cluster does, and with three databases over four nodes the per-node ceiling binds first and legitimately spreads a group's third partition to balance load. That is the intended trade, so asserting against it here would be asserting the wrong thing. --- .../blue_green_version_bump_assignment.cs | 232 ++++++++++++++++++ 1 file changed, 232 insertions(+) create mode 100644 src/Persistence/MartenTests/MultiTenancy/blue_green_version_bump_assignment.cs diff --git a/src/Persistence/MartenTests/MultiTenancy/blue_green_version_bump_assignment.cs b/src/Persistence/MartenTests/MultiTenancy/blue_green_version_bump_assignment.cs new file mode 100644 index 000000000..a2424af7c --- /dev/null +++ b/src/Persistence/MartenTests/MultiTenancy/blue_green_version_bump_assignment.cs @@ -0,0 +1,232 @@ +using IntegrationTests; +using JasperFx.Events; +using JasperFx.Events.Projections; +using Marten; +using Marten.Events.Aggregation; +using Npgsql; +using Shouldly; +using JasperFx; +using Weasel.Postgresql; +using Weasel.Postgresql.Migrations; +using Wolverine.Runtime.Agents; + +namespace MartenTests.MultiTenancy; + +/// +/// End-to-end cover for a projection version bump on a multi-database store, through the real +/// rather than +/// +/// directly — so it also covers the store-cardinality pass selection and +/// RetireSupersededAgentsAsync, both of which sit between a leader and the placement. +/// +/// The scenario is a blue/green rollout: the "blue" fleet runs the store as it is deployed today +/// and is already running its agents; the "green" fleet runs a build where one projection's +/// Version is one higher. Both fleets are the same application over the same tenant databases, +/// and the leader evaluating the assignments is a blue node — which is what makes the green fleet's +/// agents reachable only through its persisted node capabilities. +/// +public class blue_green_version_bump_assignment : IAsyncLifetime +{ + private const uint PreviousVersion = 2; + private const uint BumpedVersion = 3; + + private readonly List _tenants = ["bgtenant1", "bgtenant2", "bgtenant3"]; + + private DocumentStore _blue = null!; + private DocumentStore _green = null!; + private IReadOnlyList _blueAgents = null!; + private IReadOnlyList _greenAgents = null!; + + public async ValueTask InitializeAsync() + { + await using var conn = new NpgsqlConnection(Servers.PostgresConnectionString); + await conn.OpenAsync(TestContext.Current.CancellationToken); + + var connectionStrings = new List(); + foreach (var tenant in _tenants) + { + connectionStrings.Add(await CreateDatabaseIfNotExists(conn, tenant)); + } + + await conn.CloseAsync(); + + _blue = StoreFor(PreviousVersion, connectionStrings); + _green = StoreFor(BumpedVersion, connectionStrings); + + _blueAgents = await AdvertisedAgentsAsync(_blue); + _greenAgents = await AdvertisedAgentsAsync(_green); + } + + public async ValueTask DisposeAsync() + { + await _blue.DisposeAsync(); + await _green.DisposeAsync(); + } + + [Fact] + public async Task the_two_fleets_advertise_the_bumped_projection_under_disjoint_identities() + { + _blueAgents.ShouldNotBeEmpty(); + _greenAgents.ShouldNotBeEmpty(); + + var blueBumped = AgentsFor(_blueAgents, "Trip"); + var greenBumped = AgentsFor(_greenAgents, "Trip"); + + blueBumped.ShouldNotBeEmpty(); + greenBumped.Count.ShouldBe(blueBumped.Count); + blueBumped.Intersect(greenBumped).ShouldBeEmpty( + "a version bump has to change the agent identity, or the leader has no way to tell the two fleets apart"); + + // The projection that was NOT bumped is one and the same agent on both fleets, which is what + // makes it declared by every node and therefore free to land anywhere. + AgentsFor(_greenAgents, "Passenger") + .OrderBy(x => x.ToString()) + .ShouldBe(AgentsFor(_blueAgents, "Passenger").OrderBy(x => x.ToString())); + } + + [Fact] + public async Task the_bumped_version_is_assigned_to_the_fleet_that_declares_it() + { + var (grid, blue, green) = BuildGrid(); + + // The leader is a blue node: its own family enumerates only the previous version, exactly as in + // production, so the green agents reach the grid solely through node capabilities. + await using var leader = new EventSubscriptionAgentFamily([_blue], []); + await leader.EvaluateAssignmentsAsync(grid); + + foreach (var uri in AgentsFor(_greenAgents, "Trip")) + { + var host = grid.AgentFor(uri).AssignedNode; + host.ShouldNotBeNull($"{uri} was left unassigned, so nothing would build the new version"); + green.ShouldContain(host!, + $"{uri} may only run on a node that declares it — BuildAgentAsync throws 'Unknown event projection or subscription' on the other fleet"); + } + + foreach (var uri in AgentsFor(_blueAgents, "Trip")) + { + var host = grid.AgentFor(uri).AssignedNode; + host.ShouldNotBeNull($"{uri} was left unassigned, so the fleet still serving would stop projecting"); + blue.ShouldContain(host!, $"{uri} is the version the blue fleet serves and must stay there"); + } + } + + // The "one owner per version per database" bound that the sibling-partition preference exists for is + // asserted in CoreTests.Runtime.Agents.distribute_by_group_affinity instead. It needs agents to + // outnumber nodes the way they do in a real cluster (~5k agents over ~10 nodes); with three databases + // over four nodes the per-node ceiling binds first and legitimately spreads a group's third partition + // to balance load, which is the intended trade and not what this test is about. + + private (AssignmentGrid Grid, List Blue, List Green) BuildGrid() + { + var grid = new AssignmentGrid(); + + var blue = new List + { + grid.WithNode(1, Guid.NewGuid()).HasCapabilities(_blueAgents), + grid.WithNode(2, Guid.NewGuid()).HasCapabilities(_blueAgents) + }; + + var green = new List + { + grid.WithNode(3, Guid.NewGuid()).HasCapabilities(_greenAgents), + grid.WithNode(4, Guid.NewGuid()).HasCapabilities(_greenAgents) + }; + + // Blue is already running what it advertises. That incumbency is the whole point: it is what + // keeps a blue node in the candidate set of a group it can only partly run. + for (var i = 0; i < blue.Count; i++) + { + blue[i].Running(_blueAgents.Where((_, index) => index % blue.Count == i).ToArray()); + } + + // Mirrors NodeAgentController.EvaluateAssignmentsAsync, which seeds the grid from the union of + // every node's persisted capabilities before handing it to the families. + grid.WithAgents(_blueAgents.Concat(_greenAgents).Distinct().ToArray()); + + return (grid, blue, green); + } + + private static async Task> AdvertisedAgentsAsync(IEventStore store) + { + await using var family = new EventSubscriptionAgentFamily([store], []); + return await family.SupportedAgentsAsync(); + } + + private static DocumentStore StoreFor(uint tripVersion, IReadOnlyList connectionStrings) + { + return DocumentStore.For(opts => + { + opts.DatabaseSchemaName = "bluegreen"; + + // Nothing here writes or projects; the test only asks the store what it would advertise. + opts.AutoCreateSchemaObjects = AutoCreate.None; + + opts.MultiTenantedDatabases(tenancy => + { + for (var i = 0; i < connectionStrings.Count; i++) + { + tenancy.AddSingleTenantDatabase(connectionStrings[i], $"bgtenant{i + 1}"); + } + }); + + opts.Projections.Add(new TripProjection { Version = tripVersion }, ProjectionLifecycle.Async); + opts.Projections.Add(new PassengerProjection(), ProjectionLifecycle.Async); + }); + } + + private static async Task CreateDatabaseIfNotExists(NpgsqlConnection conn, string databaseName) + { + var builder = new NpgsqlConnectionStringBuilder(Servers.PostgresConnectionString); + + if (!await conn.DatabaseExists(databaseName)) + { + await new DatabaseSpecification().BuildDatabase(conn, databaseName); + } + + builder.Database = databaseName; + return builder.ConnectionString; + } + + private static List AgentsFor(IEnumerable agents, string projectionName) => + agents.Where(uri => uri.Segments.Any(segment => + segment.Trim('/').Equals(projectionName, StringComparison.OrdinalIgnoreCase))).ToList(); + + // The (type, name, databaseId) prefix of an agent URI, mirroring the internal + // EventSubscriptionAgentFamily.DatabaseKeyOf that group affinity keys on. + private static string DatabaseKeyOf(Uri uri) => + uri.Segments.Length >= 3 + ? $"{uri.Host}/{uri.Segments[1].Trim('/')}/{uri.Segments[2].Trim('/')}" + : uri.AbsoluteUri; +} + +public record TripStarted(Guid Id, string Description); + +public record PassengerBoarded(Guid TripId, string Name); + +public class BlueGreenTrip +{ + public Guid Id { get; set; } + public string Description { get; set; } = string.Empty; +} + +public class BlueGreenPassengerCount +{ + public Guid Id { get; set; } + public int Count { get; set; } +} + +public partial class TripProjection : SingleStreamProjection +{ + public TripProjection() => Name = "Trip"; + + public static BlueGreenTrip Create(TripStarted started) => + new() { Id = started.Id, Description = started.Description }; +} + +public partial class PassengerProjection : SingleStreamProjection +{ + public PassengerProjection() => Name = "Passenger"; + + public static BlueGreenPassengerCount Create(PassengerBoarded boarded) => + new() { Id = boarded.TripId, Count = 1 }; +} From bd513eba12b414854fb3d20f119c97355225e662 Mon Sep 17 00:00:00 2001 From: Anne Erdtsieck Date: Mon, 3 Aug 2026 13:21:58 +0200 Subject: [PATCH 4/5] camelCase the two new private helpers MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit House style for private methods and local functions in this code — canSelfHeal in NodeAgentController, and countOn a few lines up in this same file. Copilot caught the static method; the local function had the same problem. --- .../Runtime/Agents/AssignmentGrid.Distribution.cs | 14 +++++++------- 1 file changed, 7 insertions(+), 7 deletions(-) diff --git a/src/Wolverine/Runtime/Agents/AssignmentGrid.Distribution.cs b/src/Wolverine/Runtime/Agents/AssignmentGrid.Distribution.cs index aea0a7a08..557764fcb 100644 --- a/src/Wolverine/Runtime/Agents/AssignmentGrid.Distribution.cs +++ b/src/Wolverine/Runtime/Agents/AssignmentGrid.Distribution.cs @@ -189,7 +189,7 @@ public void DistributeByGroupAffinity(string scheme, Func groupKey, var partitions = sameCapabilities ? [group.ToList()] : group - .GroupBy(CapabilityKey) + .GroupBy(capabilityKey) .OrderByDescending(partition => partition.Count()) .ThenBy(partition => partition.Key, StringComparer.Ordinal) .Select(partition => partition.ToList()) @@ -243,7 +243,7 @@ public void DistributeByGroupAffinity(string scheme, Func groupKey, if (member.AssignedNode != null) { load[member.AssignedNode] = load.GetValueOrDefault(member.AssignedNode) + 1; - Remember(siblingHosts, member.AssignedNode); + remember(siblingHosts, member.AssignedNode); continue; } @@ -260,7 +260,7 @@ public void DistributeByGroupAffinity(string scheme, Func groupKey, candidate.Assign(member); load[candidate] += 1; - Remember(siblingHosts, candidate); + remember(siblingHosts, candidate); } continue; @@ -281,7 +281,7 @@ public void DistributeByGroupAffinity(string scheme, Func groupKey, load[incumbent] + members.Count <= maximum) { load[incumbent] += members.Count; - Remember(siblingHosts, incumbent); + remember(siblingHosts, incumbent); continue; } @@ -307,11 +307,11 @@ public void DistributeByGroupAffinity(string scheme, Func groupKey, } load[node] += members.Count; - Remember(siblingHosts, node); + remember(siblingHosts, node); } } - static void Remember(List hosts, Node node) + static void remember(List hosts, Node node) { if (!hosts.Contains(node)) { @@ -327,7 +327,7 @@ static void Remember(List hosts, Node node) /// blue/green split. An agent no node declares gets the empty key, so those stay together and keep the /// GH-3341 whole-group rescue. /// - private static string CapabilityKey(Agent agent) => + private static string capabilityKey(Agent agent) => string.Join(",", agent.CandidateNodes.Select(n => n.AssignedId).OrderBy(id => id)); public bool AllNodesHaveSameCapabilities(string scheme) From 2ea368f09ef6d66564923246fcf15261c0029666 Mon Sep 17 00:00:00 2001 From: Anne Erdtsieck Date: Mon, 3 Aug 2026 19:09:25 +0200 Subject: [PATCH 5/5] Model the sharded, tenant-partitioned shape in the end-to-end test MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The first version used AddSingleTenantDatabase — one tenant per database — which is not the shape this bug lives in. With one tenant per database a group is two or three agents deep, so it exercises the version split but not the thing group affinity exists for: several tenants sharing a shard database, with events partitioned per tenant, which is what makes IEventStore.DistributesAgentsPerTenant true and produces an agent per (database, tenant, projection version). Now three databases carry three tenants each with Conjoined event tenancy and UseTenantPartitionedEvents, so every database's group is nine agents across three capability classes instead of three. The failing assertion on main now reads event-subscriptions://marten/main/localhost.bgshard1/trip/all/v3/bgshard1-alpha may only run on a node that declares it which is the real URI grammar down to the tenant segment. Also asserts the shape itself. Without that, a change that stopped distributing per tenant would shrink every group back to one agent per projection and this file would keep passing while testing something much weaker. --- .../blue_green_version_bump_assignment.cs | 38 ++++++++++++++++--- 1 file changed, 32 insertions(+), 6 deletions(-) diff --git a/src/Persistence/MartenTests/MultiTenancy/blue_green_version_bump_assignment.cs b/src/Persistence/MartenTests/MultiTenancy/blue_green_version_bump_assignment.cs index a2424af7c..5ce04f44a 100644 --- a/src/Persistence/MartenTests/MultiTenancy/blue_green_version_bump_assignment.cs +++ b/src/Persistence/MartenTests/MultiTenancy/blue_green_version_bump_assignment.cs @@ -30,7 +30,11 @@ public class blue_green_version_bump_assignment : IAsyncLifetime private const uint PreviousVersion = 2; private const uint BumpedVersion = 3; - private readonly List _tenants = ["bgtenant1", "bgtenant2", "bgtenant3"]; + // Our shape, and the one that makes a group big enough to matter: several tenants share a shard + // database, and events are tenant-partitioned, so the daemon runs an agent per + // (database, tenant, projection version) rather than one per (database, version). + private static readonly string[] _databases = ["bgshard1", "bgshard2", "bgshard3"]; + private static readonly string[] _tenantsPerDatabase = ["alpha", "beta", "gamma"]; private DocumentStore _blue = null!; private DocumentStore _green = null!; @@ -43,9 +47,9 @@ public async ValueTask InitializeAsync() await conn.OpenAsync(TestContext.Current.CancellationToken); var connectionStrings = new List(); - foreach (var tenant in _tenants) + foreach (var database in _databases) { - connectionStrings.Add(await CreateDatabaseIfNotExists(conn, tenant)); + connectionStrings.Add(await CreateDatabaseIfNotExists(conn, database)); } await conn.CloseAsync(); @@ -66,8 +70,18 @@ public async ValueTask DisposeAsync() [Fact] public async Task the_two_fleets_advertise_the_bumped_projection_under_disjoint_identities() { - _blueAgents.ShouldNotBeEmpty(); - _greenAgents.ShouldNotBeEmpty(); + // Guard the shape this test exists for: agents are per (database, tenant, projection), so each + // shard database's group is tenants x projections deep. Without this, a change that stopped + // distributing per tenant would silently shrink every group to one agent per projection and this + // whole file would go on passing while testing something much weaker. + foreach (var database in _blueAgents.GroupBy(DatabaseKeyOf)) + { + database.Count().ShouldBe(_tenantsPerDatabase.Length * 2, + $"{database.Key} must carry an agent per tenant for each of the two projections"); + } + + _blueAgents.Count.ShouldBe(_databases.Length * _tenantsPerDatabase.Length * 2); + _greenAgents.Count.ShouldBe(_blueAgents.Count); var blueBumped = AgentsFor(_blueAgents, "Trip"); var greenBumped = AgentsFor(_greenAgents, "Trip"); @@ -161,14 +175,26 @@ private static DocumentStore StoreFor(uint tripVersion, IReadOnlyList co // Nothing here writes or projects; the test only asks the store what it would advertise. opts.AutoCreateSchemaObjects = AutoCreate.None; + // Sharded shape: every database holds several tenants, and the event tables are partitioned + // per tenant, which is what turns IEventStore.DistributesAgentsPerTenant on and produces the + // per-(database, tenant) agents a real shard database's group is made of. + opts.Events.TenancyStyle = JasperFx.MultiTenancy.TenancyStyle.Conjoined; + opts.Events.UseTenantPartitionedEvents = true; + opts.Events.UseArchivedStreamPartitioning = false; + opts.MultiTenantedDatabases(tenancy => { for (var i = 0; i < connectionStrings.Count; i++) { - tenancy.AddSingleTenantDatabase(connectionStrings[i], $"bgtenant{i + 1}"); + tenancy.AddMultipleTenantDatabase(connectionStrings[i], _databases[i]) + .ForTenants(_tenantsPerDatabase.Select(t => $"{_databases[i]}-{t}").ToArray()); } }); + // Conjoined events require conjoined read models. + opts.Schema.For().MultiTenanted(); + opts.Schema.For().MultiTenanted(); + opts.Projections.Add(new TripProjection { Version = tripVersion }, ProjectionLifecycle.Async); opts.Projections.Add(new PassengerProjection(), ProjectionLifecycle.Async); });