Skip to content

Pin the group-affinity invariants GH-3792 relies on, and simulate a version bump with slow starts #3804

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 1 commit into from
Aug 3, 2026
+277 −4
Merged

Pin the group-affinity invariants GH-3792 relies on, and simulate a version bump with slow starts #3804

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
145 changes: 145 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 @@ -269,6 +269,151 @@ public void a_version_bump_splits_a_group_between_the_old_and_new_version_nodes(
"the new version's agent may only run on the node that declares it — the blue node cannot build it");
}

[Fact]
public void a_settled_blue_green_split_does_not_churn_on_the_next_evaluation()
{
// GH-3785 counted ~45,000 ReassignAgent decisions in six minutes during a rollout ramp. Whatever
// else contributes to that, the placement itself must be a fixed point: re-evaluating the settled
// split state — same fleet, same capabilities, everything running exactly where the previous
// evaluation put it — must move nothing. The partition incumbent rule (members[0].AssignedNode,
// all members on one node) is subtle enough to regress silently, so pin it.
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 ids = Enumerable.Range(0, 4).Select(_ => Guid.NewGuid()).ToArray();

AssignmentGrid BuildGrid(out Dictionary<Guid, AssignmentGrid.Node> byId)
{
var grid = new AssignmentGrid();
var blues = new[] { grid.WithNode(1, ids[0]), grid.WithNode(2, ids[1]) };
var greens = new[] { grid.WithNode(3, ids[2]), grid.WithNode(4, ids[3]) };
foreach (var blue in blues) blue.HasCapabilities(previous.Concat(unchanged));
foreach (var green in greens) green.HasCapabilities(bumped.Concat(unchanged));
byId = blues.Concat(greens).ToDictionary(n => n.NodeId);
return grid;
}

var first = BuildGrid(out _);
first.WithAgents(previous.Concat(bumped).Concat(unchanged).ToArray());
first.DistributeByGroupAffinity("event-subscriptions", DatabaseKey);

var placement = first.AllAgents.ToDictionary(a => a.Uri, a => a.AssignedNode!.NodeId);

// Second evaluation: the first one's outcome is now reality — every agent RUNNING where it landed.
var second = BuildGrid(out var nodesById);
foreach (var byNode in placement.GroupBy(kv => kv.Value))
{
nodesById[byNode.Key].Running(byNode.Select(kv => kv.Key).ToArray());
}

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

var moved = second.AllAgents
.Where(a => a.AssignedNode!.NodeId != placement[a.Uri])
.Select(a => a.Uri.ToString())
.ToList();

moved.ShouldBeEmpty("a settled split state must be a fixed point of the placement, not churn");
}

[Fact]
public void a_rolling_restart_keeps_every_database_on_at_most_two_hosts()
{
// The partition key is the EXACT set of declaring node ids — stricter than "a common capable node
// exists", and flagged in the PR as a deliberate judgement call. This pins the invariant that makes
// the strictness safe to keep (or later relax): in a rolling restart, where capability sets overlap
// but are NOT equal — node 1 still on the old build and running everything, nodes 2 and 3 already
// restarted onto the new one — a database still lands on at most two hosts, so the strict key costs
// no extra connection pools. Anyone coarsening the key can refactor against this safely.
var databases = new[] { "db1", "db2", "db3" };
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 old = grid.WithNode(1, Guid.NewGuid()).HasCapabilities(previous.Concat(unchanged));
old.Running(previous.Concat(unchanged).ToArray()); // was the whole cluster before the roll began
grid.WithNode(2, Guid.NewGuid()).HasCapabilities(bumped.Concat(unchanged));
grid.WithNode(3, Guid.NewGuid()).HasCapabilities(bumped.Concat(unchanged));

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

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

grid.AllAgents.ShouldAllBe(a => a.AssignedNode != null);

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.ShouldBeLessThanOrEqualTo(2,
$"{db} landed on {hosts.Count} hosts — a rolling restart must not cost a third pool set per database");
}
}

[Fact]
public void asymmetric_fleets_strand_nothing_and_load_every_capable_node()
{
// A brand-new green fleet that declares ONLY the bumped agents (it has not been anything else yet)
// joins a blue fleet running everything. The per-node ceiling is computed over ALL nodes — including
// the ones incapable of most of the work — so this shape is where ceiling arithmetic would go wrong
// first: nothing may be left unassigned, every bumped agent must land green, and neither green node
// may sit idle while its twin takes the whole new version.
var databases = Enumerable.Range(1, 8).Select(i => $"db{i}").ToArray();
var tenants = new[] { "t1", "t2", "t3" };

Uri[] Across(uint version) =>
databases.SelectMany(db => tenants.Select(t => VersionedAgent(db, version, t))).ToArray();

var previous = Across(22);
var bumped = Across(23);

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);
blues[0].Running(previous.Take(previous.Length / 2).ToArray());
blues[1].Running(previous.Skip(previous.Length / 2).ToArray());
foreach (var green in greens) green.HasCapabilities(bumped);

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

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

grid.AllAgents.ShouldAllBe(a => a.AssignedNode != null);

foreach (var uri in bumped)
{
greens.ShouldContain(grid.AgentFor(uri).AssignedNode!,
$"{uri} may only run on the fleet that declares it");
}

greens.Select(g => g.ForScheme("event-subscriptions").Count())
.ShouldAllBe(count => count > 0, "no green node may sit idle while its twin hosts the whole new version");
}

[Fact]
public void a_split_group_costs_only_as_many_nodes_as_the_capability_split_forces()
{
Expand Down
108 changes: 105 additions & 3 deletions src/Testing/CoreTests/Runtime/Agents/slow_agent_start_convergence.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -194,6 +194,88 @@ public async Task the_unconfirmed_remainder_of_a_partially_started_chunk_still_c
cluster.StopsEmitted.ShouldBe(0);
}

/// <summary>
/// GH-3753 is not about slow starts in general — production converges fine without a version bump. It is
/// about a deploy that CONTAINS one, which means slow starts and a blue/green capability split at the same
/// time. Until GH-3792 those two conditions were each covered by tests that passed while their
/// intersection failed, so this drives the field's actual deploy shape end to end: a sharded store's
/// grouped agents (DistributeByGroupAffinity, as EventSubscriptionAgentFamily uses for a multi-database
/// store), a green fleet that alone declares the bumped version, a blue leader whose own family cannot even
/// enumerate the green agents — and the long-tailed start costs on top.
/// </summary>
[Fact]
public async Task a_version_bump_converges_with_slow_starts_and_no_cross_fleet_placement()
{
var databases = Enumerable.Range(1, 12).Select(i => $"db{i:D2}").ToArray();
var tenants = new[] { "t1", "t2", "t3" };

string[] Names(string kind) =>
databases.SelectMany(db => tenants.Select(t => $"{db}/{kind}/{t}")).ToArray();

var previous = Names("v22"); // built only by the blue fleet
var bumped = Names("v23"); // built only by the green fleet
var unchanged = Names("same"); // the projections whose version did not change — built by everyone

Uri[] Uris(IEnumerable<string> names) => names.Select(x => new Uri($"fake://{x}")).ToArray();

// The leader is blue: exactly as in production, its own family enumerates only what its store
// registers, so the green agents reach the grid solely through the green nodes' capabilities.
var family = new FakeAgentFamily("fake", previous.Concat(unchanged).ToArray())
{
// What EventSubscriptionAgentFamily.EvaluateAssignmentsAsync does for a multi-database store:
// group affinity keyed on the database segment.
Distribution = grid => grid.DistributeByGroupAffinity("fake", uri => uri.Host)
};

var blueCapabilities = Uris(previous.Concat(unchanged));
var greenCapabilities = Uris(bumped.Concat(unchanged));

// Nodes 0-1 blue (node 0 is the leader), nodes 2-3 green.
var cluster = new SlowStartCluster(nodeCount: 4, family, seed: 3753,
capabilitiesFor: i => i < 2 ? blueCapabilities : greenCapabilities);
cluster.StartCost = fieldStartCost(3753, cluster.AllAgents);

var rounds = await cluster.RunUntilConvergedAsync(maxRounds: 60);

// Every agent of BOTH fleets is running — the bug behind GH-3792 made this impossible: the bumped
// agents were assigned to blue nodes that cannot build them and the new version never started.
cluster.RunningAgents.Count.ShouldBe(cluster.AllAgents.Length);
rounds.ShouldBeLessThan(60);

// No agent ever landed on a fleet that cannot build it, at any point during the wave.
var blues = new[] { cluster.NodeIdAt(0), cluster.NodeIdAt(1) };
var greens = new[] { cluster.NodeIdAt(2), cluster.NodeIdAt(3) };

foreach (var uri in Uris(previous))
{
blues.ShouldContain(cluster.RunningAssignments[uri], $"{uri} is the blue fleet's version");
}

foreach (var uri in Uris(bumped))
{
greens.ShouldContain(cluster.RunningAssignments[uri], $"{uri} is the green fleet's version");
}

// The slow wave generated no churn: nothing was stopped or re-decided while starts were in flight.
cluster.DoubleStartReports.ShouldBeEmpty();
cluster.StopsEmitted.ShouldBe(0);
cluster.ReassignmentsEmitted.ShouldBe(0);

// And the connection-pool bound that group affinity exists for held through the whole rollout shape:
// a database's agents sit on at most two nodes — one per version — not one per agent.
foreach (var db in databases)
{
var hosts = tenants
.SelectMany(t => new[] { $"{db}/v22/{t}", $"{db}/v23/{t}", $"{db}/same/{t}" })
.Select(name => cluster.RunningAssignments[new Uri($"fake://{name}")])
.Distinct()
.ToList();

hosts.Count.ShouldBeLessThanOrEqualTo(2,
$"{db} is hosted by {hosts.Count} nodes — a version bump must cost one owner per version, not a pool set per partition");
}
}

/// <summary>
/// A simulated multi-node cluster driving the leader's real assignment evaluation. One <see cref="RunRoundAsync" />
/// is one health-check tick: in-flight starts advance, the leader evaluates, and the commands it emits are
Expand Down Expand Up @@ -222,6 +304,19 @@ private sealed class SlowStartCluster
private record struct InFlightStart(Guid NodeId, int RoundsRemaining);

public SlowStartCluster(int nodeCount, int agentCount, int seed)
: this(nodeCount, new FakeAgentFamily("fake", agentCount), seed)
{
}

/// <summary>
/// The seams a blue/green scenario needs: the leader's own <paramref name="family" /> (which, as in
/// production, may enumerate only its OWN fleet's agents), and per-node capability sets via
/// <paramref name="capabilitiesFor" /> (node index -> declared agents). The other fleet's agents
/// reach the leader's grid exactly the way they do for real — through the capability union
/// (NodeAgentController.EvaluateAssignmentsAsync seeds the grid from every node's capabilities).
/// </summary>
public SlowStartCluster(int nodeCount, FakeAgentFamily family, int seed,
Func<int, Uri[]>? capabilitiesFor = null)
{
Options = new WolverineOptions { ApplicationAssembly = GetType().Assembly };
Options.Transports.NodeControlEndpoint = new FakeEndpoint("fake://self".ToUri(), EndpointRole.System);
Expand All @@ -232,8 +327,10 @@ public SlowStartCluster(int nodeCount, int agentCount, int seed)
_runtime.DurabilitySettings.Returns(Options.Durability);
_runtime.Observer.Returns(Substitute.For<IWolverineObserver>());

_family = new FakeAgentFamily("fake", agentCount);
AllAgents = _family.AllAgentUris();
_family = family;
AllAgents = capabilitiesFor is null
? _family.AllAgentUris()
: Enumerable.Range(0, nodeCount).SelectMany(capabilitiesFor).Distinct().ToArray();

_controller = new NodeAgentController(_runtime, Substitute.For<INodeAgentPersistence>(), [_family],
NullLogger<NodeAgentController>.Instance, CancellationToken.None);
Expand Down Expand Up @@ -263,7 +360,7 @@ public SlowStartCluster(int nodeCount, int agentCount, int seed)
ControlUri = new Uri($"fake://node{i}")
};

node.Capabilities.AddRange(AllAgents);
node.Capabilities.AddRange(capabilitiesFor?.Invoke(i) ?? AllAgents);
_nodes.Add(node);
}

Expand All @@ -286,6 +383,11 @@ public SlowStartCluster(int nodeCount, int agentCount, int seed)

public IReadOnlyList<Uri> RunningAgents => _running.Keys.ToList();

/// <summary>Ground truth of where each agent is actually running, for placement assertions.</summary>
public IReadOnlyDictionary<Uri, Guid> RunningAssignments => _running;

public Guid NodeIdAt(int index) => _nodes[index].NodeId;

public int[] RunningCountsByNode
=> _nodes.Select(node => _running.Count(x => x.Value == node.NodeId)).ToArray();

Expand Down
28 changes: 27 additions & 1 deletion src/Testing/Wolverine.ComplianceTests/FakeAgentFamily.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -37,6 +37,16 @@ public FakeAgentFamily(string scheme, int agentCount)
AgentNames = Enumerable.Range(0, agentCount).Select(i => $"agent-{i:D5}").ToArray();
}

/// <summary>
/// A family over explicit agent names — names may carry path segments (e.g. <c>db01/v22/t1</c>) so a
/// test can model grouped, versioned agents the way <c>EventSubscriptionAgentFamily</c> URIs do.
/// </summary>
public FakeAgentFamily(string scheme, IReadOnlyList<string> agentNames)
{
Scheme = scheme;
AgentNames = agentNames;
}

public string Scheme { get; } = "fake";

public static string[] Names =
Expand Down Expand Up @@ -71,9 +81,25 @@ public FakeAgentFamily(string scheme, int agentCount)

public LightweightCache<Uri, FakeAgent> Agents { get; } = new(x => new FakeAgent(x));

/// <summary>
/// Override how this family distributes its agents over the grid. Defaults to the capability-blind
/// <see cref="AssignmentGrid.DistributeEvenly(string)" />; a test simulating a sharded store sets this to
/// <see cref="AssignmentGrid.DistributeByGroupAffinity(string, Func{Uri, string})" /> the way
/// <c>EventSubscriptionAgentFamily</c> does for a multi-database store.
/// </summary>
public Action<AssignmentGrid>? Distribution { get; set; }

public ValueTask EvaluateAssignmentsAsync(AssignmentGrid assignments)
{
assignments.DistributeEvenly(Scheme);
if (Distribution != null)
{
Distribution(assignments);
}
else
{
assignments.DistributeEvenly(Scheme);
}

return new ValueTask();
}

Expand Down
Loading