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 f099b178a..86b497a75 100644 --- a/src/Testing/CoreTests/Runtime/Agents/distribute_by_group_affinity.cs +++ b/src/Testing/CoreTests/Runtime/Agents/distribute_by_group_affinity.cs @@ -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 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 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 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() { diff --git a/src/Testing/CoreTests/Runtime/Agents/slow_agent_start_convergence.cs b/src/Testing/CoreTests/Runtime/Agents/slow_agent_start_convergence.cs index a423fab38..e27d2e6bc 100644 --- a/src/Testing/CoreTests/Runtime/Agents/slow_agent_start_convergence.cs +++ b/src/Testing/CoreTests/Runtime/Agents/slow_agent_start_convergence.cs @@ -194,6 +194,88 @@ public async Task the_unconfirmed_remainder_of_a_partially_started_chunk_still_c cluster.StopsEmitted.ShouldBe(0); } + /// + /// 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. + /// + [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 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"); + } + } + /// /// A simulated multi-node cluster driving the leader's real assignment evaluation. One /// is one health-check tick: in-flight starts advance, the leader evaluates, and the commands it emits are @@ -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) + { + } + + /// + /// The seams a blue/green scenario needs: the leader's own (which, as in + /// production, may enumerate only its OWN fleet's agents), and per-node capability sets via + /// (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). + /// + public SlowStartCluster(int nodeCount, FakeAgentFamily family, int seed, + Func? capabilitiesFor = null) { Options = new WolverineOptions { ApplicationAssembly = GetType().Assembly }; Options.Transports.NodeControlEndpoint = new FakeEndpoint("fake://self".ToUri(), EndpointRole.System); @@ -232,8 +327,10 @@ public SlowStartCluster(int nodeCount, int agentCount, int seed) _runtime.DurabilitySettings.Returns(Options.Durability); _runtime.Observer.Returns(Substitute.For()); - _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(), [_family], NullLogger.Instance, CancellationToken.None); @@ -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); } @@ -286,6 +383,11 @@ public SlowStartCluster(int nodeCount, int agentCount, int seed) public IReadOnlyList RunningAgents => _running.Keys.ToList(); + /// Ground truth of where each agent is actually running, for placement assertions. + public IReadOnlyDictionary RunningAssignments => _running; + + public Guid NodeIdAt(int index) => _nodes[index].NodeId; + public int[] RunningCountsByNode => _nodes.Select(node => _running.Count(x => x.Value == node.NodeId)).ToArray(); diff --git a/src/Testing/Wolverine.ComplianceTests/FakeAgentFamily.cs b/src/Testing/Wolverine.ComplianceTests/FakeAgentFamily.cs index 44a8a6105..d40d9905b 100644 --- a/src/Testing/Wolverine.ComplianceTests/FakeAgentFamily.cs +++ b/src/Testing/Wolverine.ComplianceTests/FakeAgentFamily.cs @@ -37,6 +37,16 @@ public FakeAgentFamily(string scheme, int agentCount) AgentNames = Enumerable.Range(0, agentCount).Select(i => $"agent-{i:D5}").ToArray(); } + /// + /// A family over explicit agent names — names may carry path segments (e.g. db01/v22/t1) so a + /// test can model grouped, versioned agents the way EventSubscriptionAgentFamily URIs do. + /// + public FakeAgentFamily(string scheme, IReadOnlyList agentNames) + { + Scheme = scheme; + AgentNames = agentNames; + } + public string Scheme { get; } = "fake"; public static string[] Names = @@ -71,9 +81,25 @@ public FakeAgentFamily(string scheme, int agentCount) public LightweightCache Agents { get; } = new(x => new FakeAgent(x)); + /// + /// Override how this family distributes its agents over the grid. Defaults to the capability-blind + /// ; a test simulating a sharded store sets this to + /// the way + /// EventSubscriptionAgentFamily does for a multi-database store. + /// + public Action? Distribution { get; set; } + public ValueTask EvaluateAssignmentsAsync(AssignmentGrid assignments) { - assignments.DistributeEvenly(Scheme); + if (Distribution != null) + { + Distribution(assignments); + } + else + { + assignments.DistributeEvenly(Scheme); + } + return new ValueTask(); }