diff --git a/src/Persistence/MartenTests/MultiTenancy/durability_projection_affinity_real_stores.cs b/src/Persistence/MartenTests/MultiTenancy/durability_projection_affinity_real_stores.cs new file mode 100644 index 000000000..c36f4fb84 --- /dev/null +++ b/src/Persistence/MartenTests/MultiTenancy/durability_projection_affinity_real_stores.cs @@ -0,0 +1,146 @@ +using IntegrationTests; +using JasperFx; +using JasperFx.Events.Projections; +using Marten; +using Microsoft.Extensions.Logging.Abstractions; +using Npgsql; +using Shouldly; +using Weasel.Postgresql; +using Weasel.Postgresql.Migrations; +using Wolverine; +using Wolverine.Persistence; +using Wolverine.Persistence.Durability; +using Wolverine.Postgresql; +using Wolverine.RDBMS; +using Wolverine.Runtime.Agents; + +namespace MartenTests.MultiTenancy; + +/// +/// GH-3785, the half a unit test cannot cover: the cross-family affinity joins a +/// wolverinedb:// agent URI (built by MessageDatabase from a Weasel descriptor) to an +/// event-subscriptions:// agent URI (built from a Marten database descriptor), and the two +/// families describe the same physical database through entirely different pipelines. If their server +/// or database spellings ever diverge, the join silently never engages — which looks exactly like the +/// feature working, minus the benefit. So this runs both REAL pipelines against the same three tenant +/// databases and asserts the join actually connects them. +/// +public class durability_projection_affinity_real_stores : IAsyncLifetime +{ + private static readonly string[] _databases = ["affshard1", "affshard2", "affshard3"]; + private static readonly string[] _tenantsPerDatabase = ["alpha", "beta", "gamma"]; + + private DocumentStore _martenStore = null!; + private readonly List _messageStores = []; + private IReadOnlyList _projectionAgents = 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 database in _databases) + { + var builder = new NpgsqlConnectionStringBuilder(Servers.PostgresConnectionString); + if (!await conn.DatabaseExists(database)) + { + await new DatabaseSpecification().BuildDatabase(conn, database); + } + + builder.Database = database; + connectionStrings.Add(builder.ConnectionString); + } + + await conn.CloseAsync(); + + // Pipeline one: the real Marten store advertising per-(database, tenant) projection agents. + _martenStore = DocumentStore.For(opts => + { + opts.DatabaseSchemaName = "affinity"; + opts.AutoCreateSchemaObjects = AutoCreate.None; + + 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.AddMultipleTenantDatabase(connectionStrings[i], _databases[i]) + .ForTenants(_tenantsPerDatabase.Select(t => $"{_databases[i]}-{t}").ToArray()); + } + }); + + opts.Schema.For().MultiTenanted(); + opts.Projections.Add(new TripProjection { Version = 2 }, ProjectionLifecycle.Async); + }); + + await using var family = new EventSubscriptionAgentFamily([_martenStore], []); + _projectionAgents = await family.SupportedAgentsAsync(); + + // Pipeline two: a real Wolverine Postgres message store per tenant database, whose Uri is the + // wolverinedb:// identity the durability family distributes. + foreach (var connectionString in connectionStrings) + { + var settings = new DatabaseSettings + { + ConnectionString = connectionString, + Role = MessageStoreRole.Tenant, + SchemaName = "wolverine" + }; + + _messageStores.Add(new PostgresqlMessageStore(settings, new DurabilitySettings(), + NpgsqlDataSource.Create(connectionString), NullLogger.Instance)); + } + } + + public async ValueTask DisposeAsync() + { + foreach (var store in _messageStores) + { + await store.DisposeAsync(); + } + + _martenStore.Dispose(); + } + + [Fact] + public void the_real_uri_pipelines_join_and_every_database_co_locates() + { + _projectionAgents.Count.ShouldBe(_databases.Length * _tenantsPerDatabase.Length, + "an agent per (database, tenant) is the shape that makes this worth testing"); + + var grid = new AssignmentGrid(); + grid.WithNode(1, Guid.NewGuid()); + grid.WithNode(2, Guid.NewGuid()); + + var durabilityAgents = _messageStores.Select(x => x.Uri).ToArray(); + grid.WithAgents(_projectionAgents.Concat(durabilityAgents).ToArray()); + + // The two passes in the order NodeAgentController guarantees, both over real URIs. + grid.DistributeByGroupAffinity(EventSubscriptionAgentFamily.SchemeName, + EventSubscriptionAgentFamily.DatabaseKeyOf); + grid.DistributeEvenlyWithAffinity(PersistenceConstants.AgentScheme, + DurabilityProjectionAffinity.BuildPreference(grid)); + + grid.AllAgents.ShouldAllBe(a => a.AssignedNode != null); + + for (var i = 0; i < _databases.Length; i++) + { + var database = _databases[i]; + var projectionOwners = _projectionAgents + .Where(uri => EventSubscriptionAgentFamily.DatabaseIdOf(uri)?.Name == database) + .Select(uri => grid.AgentFor(uri).AssignedNode) + .Distinct() + .ToList(); + + projectionOwners.ShouldHaveSingleItem($"{database}'s projection agents must be on one node"); + + grid.AgentFor(durabilityAgents[i]).AssignedNode.ShouldBe(projectionOwners.Single(), + $"{database}'s durability agent must land with its projections — if this fails the two " + + "families' database spellings have diverged and the affinity join is silently dead"); + } + } +} diff --git a/src/Testing/CoreTests/Runtime/Agents/durability_follows_projection_affinity.cs b/src/Testing/CoreTests/Runtime/Agents/durability_follows_projection_affinity.cs new file mode 100644 index 000000000..83a048171 --- /dev/null +++ b/src/Testing/CoreTests/Runtime/Agents/durability_follows_projection_affinity.cs @@ -0,0 +1,278 @@ +using JasperFx.Descriptors; +using Microsoft.Extensions.Logging.Abstractions; +using NSubstitute; +using Wolverine.ComplianceTests; +using Wolverine.Runtime; +using Wolverine.Runtime.Agents; +using Xunit; + +namespace CoreTests.Runtime.Agents; + +// GH-3785: database affinity is a property of the DATABASE, not of each agent family independently. The +// event-subscription family already keeps a shard database's projection agents together on one node +// (GH-3792 / marten#4806); the durability agent for that same database must land on that node too, or the +// database attracts two nodes' connection pools instead of one — measured on a 512-shard production +// cluster as 73% of databases split, ~425 connections held by durability owners against databases they +// otherwise never touch. +public class durability_follows_projection_affinity +{ + private static string DatabaseKey(Uri uri) => uri.Segments[2].Trim('/'); + + // event-subscriptions://{type}/{name}/{databaseId}/{shard...} — the databaseId slot carries + // DatabaseId.ToString(), which is how EventSubscriptionAgentFamily.DatabaseIdOf parses it back. + private static Uri Projection(string server, string db, string tenant) => + new($"event-subscriptions://marten/main/{new DatabaseId(server, db)}/Proj/All/v7/{tenant}"); + + // wolverinedb://{engine}/{server}/{database}/{schema} — the grammar MessageDatabase builds. + private static Uri Durability(string server, string db) => + new($"wolverinedb://postgresql/{server}/{db}/wolverine"); + + private static void distribute(AssignmentGrid grid) + { + // The two passes in the order NodeAgentController now guarantees: event subscriptions first, + // durability after, over the same grid. + grid.DistributeByGroupAffinity("event-subscriptions", DatabaseKey); + grid.DistributeEvenlyWithAffinity("wolverinedb", DurabilityProjectionAffinityAccess.BuildPreference(grid)); + } + + [Fact] + public void a_databases_durability_agent_lands_with_its_projection_owner() + { + var databases = new[] { "tenant1", "tenant2", "tenant3", "tenant4" }; + var tenants = new[] { "t1", "t2", "t3" }; + + var projections = databases + .SelectMany(db => tenants.Select(t => Projection("localhost", db, t))) + .ToArray(); + var durability = databases.Select(db => Durability("localhost", db)).ToArray(); + + var grid = new AssignmentGrid(); + grid.WithNode(1, Guid.NewGuid()); + grid.WithNode(2, Guid.NewGuid()); + grid.WithAgents(projections.Concat(durability).ToArray()); + + distribute(grid); + + foreach (var db in databases) + { + var projectionOwner = grid.AgentFor(Projection("localhost", db, "t1")).AssignedNode; + projectionOwner.ShouldNotBeNull(); + + grid.AgentFor(Durability("localhost", db)).AssignedNode.ShouldBe(projectionOwner, + $"{db}'s durability agent must live with {db}'s projections so the database attracts one pool, not two"); + } + } + + [Fact] + public void a_durability_agent_running_elsewhere_is_moved_to_the_projection_owner() + { + // The one-time migration that converges an existing cluster: the durability agent is RUNNING on + // node 2, the projections land on node 1, and the agent must move — surfacing as a normal + // ReassignAgent command downstream. + var projections = new[] { "t1", "t2" }.Select(t => Projection("localhost", "tenant1", t)).ToArray(); + var durabilityUri = Durability("localhost", "tenant1"); + + var grid = new AssignmentGrid(); + var node1 = grid.WithNode(1, Guid.NewGuid()); + node1.Running(projections); + var node2 = grid.WithNode(2, Guid.NewGuid()); + node2.Running(durabilityUri); + + distribute(grid); + + grid.AgentFor(durabilityUri).AssignedNode.ShouldBe(node1, + "the durability agent follows the projections even when it means moving a running agent once"); + } + + [Fact] + public void the_settled_co_located_state_is_a_fixed_point() + { + // Once durability and projections share a node, re-evaluating must move nothing — churn here is + // agent restarts at 512-database scale. + var databases = new[] { "tenant1", "tenant2", "tenant3", "tenant4" }; + var tenants = new[] { "t1", "t2" }; + + var first = new AssignmentGrid(); + first.WithNode(1, Guid.NewGuid()); + first.WithNode(2, Guid.NewGuid()); + first.WithAgents(databases + .SelectMany(db => tenants.Select(t => Projection("localhost", db, t)).Append(Durability("localhost", db))) + .ToArray()); + + distribute(first); + var placement = first.AllAgents.ToDictionary(a => a.Uri, a => a.AssignedNode!.NodeId); + + var ids = first.Nodes.ToDictionary(n => n.NodeId, n => n.AssignedId); + var second = new AssignmentGrid(); + var nodesById = ids.ToDictionary(pair => pair.Key, pair => second.WithNode(pair.Value, pair.Key)); + foreach (var byNode in placement.GroupBy(kv => kv.Value)) + { + nodesById[byNode.Key].Running(byNode.Select(kv => kv.Key).ToArray()); + } + + distribute(second); + + second.AllAgents + .Where(a => a.AssignedNode!.NodeId != placement[a.Uri]) + .ShouldBeEmpty("the co-located state must be a fixed point of both distributions"); + } + + [Fact] + public void a_database_with_no_projections_still_gets_an_even_spread() + { + // The main message store (or any database with no async projections) has nothing to follow; those + // agents keep today's even distribution and are never stranded. + var durability = Enumerable.Range(1, 6).Select(i => Durability("localhost", $"plain{i}")).ToArray(); + + var grid = new AssignmentGrid(); + grid.WithNode(1, Guid.NewGuid()); + grid.WithNode(2, Guid.NewGuid()); + grid.WithAgents(durability); + + distribute(grid); + + grid.AllAgents.ShouldAllBe(a => a.AssignedNode != null); + grid.Nodes.Select(n => n.ForScheme("wolverinedb").Count()) + .ShouldAllBe(count => count == 3, "with no affinity available the spread stays even"); + } + + [Fact] + public void the_same_database_name_on_two_servers_is_disambiguated_by_server() + { + // Two servers each carry a database literally named "app": the join must not guess. The durability + // agent for server-a's "app" follows server-a's projections, and vice versa. + var grid = new AssignmentGrid(); + var node1 = grid.WithNode(1, Guid.NewGuid()); + var node2 = grid.WithNode(2, Guid.NewGuid()); + + node1.Running(Projection("server-a", "app", "t1"), Projection("server-a", "app", "t2")); + node2.Running(Projection("server-b", "app", "t1"), Projection("server-b", "app", "t2")); + + grid.WithAgents(Durability("server-a", "app"), Durability("server-b", "app")); + + distribute(grid); + + grid.AgentFor(Durability("server-a", "app")).AssignedNode.ShouldBe(node1); + grid.AgentFor(Durability("server-b", "app")).AssignedNode.ShouldBe(node2); + } + + [Fact] + public void an_ambiguous_name_with_an_unrecognized_server_spelling_falls_back_to_even() + { + // Ambiguous name AND a server spelling matching neither candidate: affinity must decline rather + // than guess — a miss is never wrong, only not-better — and the agent still gets assigned. + var grid = new AssignmentGrid(); + var node1 = grid.WithNode(1, Guid.NewGuid()); + var node2 = grid.WithNode(2, Guid.NewGuid()); + + node1.Running(Projection("server-a", "app", "t1")); + node2.Running(Projection("server-b", "app", "t1")); + + var orphan = Durability("server-c", "app"); + grid.WithAgents(orphan); + + distribute(grid); + + grid.AgentFor(orphan).AssignedNode.ShouldNotBeNull("no affinity is no reason to strand the agent"); + } + + [Fact] + public void differing_port_spellings_still_join() + { + // The two families describe the same physical server through different pipelines, and either side + // may carry a port the other doesn't. The join has to survive that or it silently never engages — + // which would look exactly like the feature working, minus the benefit. + var grid = new AssignmentGrid(); + var node1 = grid.WithNode(1, Guid.NewGuid()); + var node2 = grid.WithNode(2, Guid.NewGuid()); + + node1.Running(Projection("db-host:5433", "app", "t1")); + node2.Running(Projection("other-host", "app", "t1")); + + var durability = Durability("db-host", "app"); + grid.WithAgents(durability); + + distribute(grid); + + grid.AgentFor(durability).AssignedNode.ShouldBe(node1, + "db-host:5433 and db-host are the same server; the port must not defeat the join"); + } + + [Fact] + public void during_a_blue_green_split_the_durability_agent_follows_the_larger_side() + { + // Mid-rollout a database legitimately has one owner per projection version. The durability agent + // follows whichever node holds more of the database's agents — deterministically, so it does not + // flap between the two owners across evaluations. + var grid = new AssignmentGrid(); + var blue = grid.WithNode(1, Guid.NewGuid()); + var green = grid.WithNode(2, Guid.NewGuid()); + + blue.Running(Projection("localhost", "tenant1", "t1"), Projection("localhost", "tenant1", "t2")); + green.Running(Projection("localhost", "tenant1", "t3")); + + var durability = Durability("localhost", "tenant1"); + grid.WithAgents(durability); + + distribute(grid); + + grid.AgentFor(durability).AssignedNode.ShouldBe(blue, + "two of tenant1's three agents are on blue, so blue is the cheaper co-location"); + } + + [Fact] + public async Task the_durability_family_always_evaluates_after_the_others() + { + // The affinity only works because the durability family sees the event-subscription assignments + // in the shared grid — which was previously true only by Dictionary insertion-order accident. + // Register the durability-schemed family FIRST and assert it still evaluates LAST. + var calls = new List(); + + IAgentFamily recording(string scheme) + { + var family = Substitute.For(); + family.Scheme.Returns(scheme); + family.AllKnownAgentsAsync().Returns(new ValueTask>(Array.Empty())); + family.EvaluateAssignmentsAsync(Arg.Any()) + .Returns(_ => + { + calls.Add(scheme); + return ValueTask.CompletedTask; + }); + return family; + } + + var options = new WolverineOptions { ApplicationAssembly = GetType().Assembly }; + options.Durability.DurabilityAgentEnabled = false; + + var runtime = Substitute.For(); + runtime.Options.Returns(options); + runtime.DurabilitySettings.Returns(options.Durability); + runtime.Observer.Returns(Substitute.For()); + + var controller = new NodeAgentController(runtime, Substitute.For(), + [recording("wolverinedb"), recording("event-subscriptions"), recording("fake")], + NullLogger.Instance, CancellationToken.None); + + var node = new WolverineNode + { + NodeId = options.UniqueNodeId, + AssignedNodeNumber = 1, + ControlUri = new Uri("fake://node0") + }; + + await controller.EvaluateAssignmentsAsync([node], new AgentRestrictions()); + + calls.Last().ShouldBe("wolverinedb", + "the durability family must evaluate after every other family so cross-family affinity can see their assignments"); + calls.Count.ShouldBe(3); + } +} + +// DurabilityProjectionAffinity is internal to Wolverine; CoreTests has InternalsVisibleTo, and this alias +// keeps the intent readable at the call sites above. +internal static class DurabilityProjectionAffinityAccess +{ + internal static Func BuildPreference(AssignmentGrid grid) + => Wolverine.Persistence.DurabilityProjectionAffinity.BuildPreference(grid); +} diff --git a/src/Wolverine/Persistence/Durability/MultiTenantedMessageDatabase.Agents.cs b/src/Wolverine/Persistence/Durability/MultiTenantedMessageDatabase.Agents.cs index d119073c0..44438b72e 100644 --- a/src/Wolverine/Persistence/Durability/MultiTenantedMessageDatabase.Agents.cs +++ b/src/Wolverine/Persistence/Durability/MultiTenantedMessageDatabase.Agents.cs @@ -47,7 +47,8 @@ public ValueTask> SupportedAgentsAsync() public ValueTask EvaluateAssignmentsAsync(AssignmentGrid assignments) { - assignments.DistributeEvenly(Scheme); + // GH-3785: same cross-family database affinity as MessageStoreCollection — see the note there. + assignments.DistributeEvenlyWithAffinity(Scheme, DurabilityProjectionAffinity.BuildPreference(assignments)); return ValueTask.CompletedTask; } diff --git a/src/Wolverine/Persistence/DurabilityProjectionAffinity.cs b/src/Wolverine/Persistence/DurabilityProjectionAffinity.cs new file mode 100644 index 000000000..87c129f65 --- /dev/null +++ b/src/Wolverine/Persistence/DurabilityProjectionAffinity.cs @@ -0,0 +1,133 @@ +using JasperFx.Descriptors; +using Wolverine.Runtime.Agents; + +namespace Wolverine.Persistence; + +/// +/// GH-3785: database affinity ACROSS agent families. The event-subscription family keeps a shard +/// database's projection agents together on one node (GH-3792 / marten#4806), but the durability agent +/// for that same database was distributed independently — so for most databases one node owned the +/// durability agent while a different node owned the projections, and the database attracted two nodes' +/// connection pools instead of one (measured on a 512-shard production cluster: 73% of databases split, +/// ~425 connections held by durability owners against databases they otherwise never touch). +/// +/// This computes, for a durability agent, the node that already owns that database's +/// event-subscription agents in the current assignment pass. It only works because +/// evaluates the durability family AFTER the event-subscription +/// family over the same shared . +/// +internal static class DurabilityProjectionAffinity +{ + /// + /// The (server, database) identity of a durability agent URI, or null for URIs that don't carry one + /// (the null store, the multi-tenanted composite marker, RavenDb-style single-segment URIs). + /// Grammar per MessageDatabase: wolverinedb://{engine}/{server}/{database}/{schema}. + /// + internal static (string Server, string Database)? DatabaseOf(Uri uri) + { + if (!uri.Scheme.Equals(PersistenceConstants.AgentScheme, StringComparison.OrdinalIgnoreCase) + || uri.Segments.Length < 3) + { + return null; + } + + return (uri.Segments[1].Trim('/'), uri.Segments[2].Trim('/')); + } + + /// + /// Build the preference function for from + /// the event-subscription assignments already present in the grid. Returns the node hosting the most + /// of a database's event-subscription agents (during a blue/green split a database legitimately has + /// one owner per version; the durability agent follows the larger side, tie-broken by node id for + /// determinism). + /// + internal static Func BuildPreference(AssignmentGrid assignments) + { + // (DatabaseId, node) -> how many of that database's event-subscription agents the node holds + var owners = new Dictionary>(); + + foreach (var agent in assignments.AgentsForScheme(EventSubscriptionAgentFamily.SchemeName)) + { + if (agent.AssignedNode == null) + { + continue; + } + + var databaseId = EventSubscriptionAgentFamily.DatabaseIdOf(agent.Uri); + if (databaseId == null) + { + continue; + } + + if (!owners.TryGetValue(databaseId, out var perNode)) + { + owners[databaseId] = perNode = new Dictionary(); + } + + perNode[agent.AssignedNode] = perNode.GetValueOrDefault(agent.AssignedNode) + 1; + } + + if (owners.Count == 0) + { + return _ => null; + } + + var ownerOf = owners.ToDictionary( + pair => pair.Key, + pair => pair.Value + .OrderByDescending(x => x.Value) + .ThenBy(x => x.Key.AssignedId) + .First().Key); + + // The durability URI spells the database as raw (server, name) segments while the + // event-subscription URI carries a DatabaseId, and the two families describe the same physical + // database through different pipelines — so join on the database NAME when that is unambiguous, + // and only fall back to comparing server spellings when two servers carry the same database name. + // A miss here is never wrong, only not-better: the agent falls back to the even spread, which is + // exactly today's behavior. + var byName = ownerOf + .GroupBy(pair => pair.Key.Name, StringComparer.OrdinalIgnoreCase) + .ToDictionary(g => g.Key, g => g.ToList(), StringComparer.OrdinalIgnoreCase); + + return uri => + { + var database = DatabaseOf(uri); + if (database == null) + { + return null; + } + + if (!byName.TryGetValue(database.Value.Database, out var candidates)) + { + return null; + } + + if (candidates.Count == 1) + { + return candidates[0].Value; + } + + var server = normalizeServer(database.Value.Server); + var matching = candidates + .Where(x => normalizeServer(x.Key.Server) == server) + .ToList(); + + return matching.Count == 1 ? matching[0].Value : null; + }; + } + + // The durability URI's server segment is descriptor.ServerName.Split(',')[0] while DatabaseId.Server + // is the unsplit descriptor value, and either side may or may not carry a port — normalize both down + // to the bare lowercase host before comparing. + private static string normalizeServer(string server) + { + var host = server.Split(',')[0]; + var colon = host.IndexOf(':'); + if (colon >= 0) + { + host = host[..colon]; + } + + return host.Trim().ToLowerInvariant(); + } +} diff --git a/src/Wolverine/Persistence/MessageStoreCollection.cs b/src/Wolverine/Persistence/MessageStoreCollection.cs index 01e74b057..422e4163e 100644 --- a/src/Wolverine/Persistence/MessageStoreCollection.cs +++ b/src/Wolverine/Persistence/MessageStoreCollection.cs @@ -317,10 +317,14 @@ public ValueTask> SupportedAgentsAsync() public ValueTask EvaluateAssignmentsAsync(AssignmentGrid assignments) { - assignments.DistributeEvenly(Scheme); + // GH-3785: a shard database's durability agent follows that database's event-subscription agents, + // so the database attracts one node's connection pool instead of two. Depends on + // NodeAgentController evaluating this family AFTER the event-subscription family. A database with + // no projection agents in the grid falls back to the even spread. + assignments.DistributeEvenlyWithAffinity(Scheme, DurabilityProjectionAffinity.BuildPreference(assignments)); return ValueTask.CompletedTask; } - + internal async Task StartScheduledJobProcessing(IWolverineRuntime runtime) { // First, find all unique message stores diff --git a/src/Wolverine/Runtime/Agents/AssignmentGrid.Distribution.cs b/src/Wolverine/Runtime/Agents/AssignmentGrid.Distribution.cs index 557764fcb..13627d7c5 100644 --- a/src/Wolverine/Runtime/Agents/AssignmentGrid.Distribution.cs +++ b/src/Wolverine/Runtime/Agents/AssignmentGrid.Distribution.cs @@ -330,6 +330,110 @@ static void remember(List hosts, Node node) private static string capabilityKey(Agent agent) => string.Join(",", agent.CandidateNodes.Select(n => n.AssignedId).OrderBy(id => id)); + /// + /// Distribute agents of a scheme evenly across the nodes — except that any agent for which + /// names a node is placed on that node regardless of the even + /// spread. Built for cross-family database affinity (GH-3785): a shard database's durability agent + /// should live on whichever node already owns that database's event-subscription agents, so the + /// database attracts ONE node's connection pool instead of two. Preferred placements are deliberately + /// not ceiling-bounded — they piggyback on the other family's own balanced distribution, and the whole + /// point is to co-locate with it even when that costs strict evenness here. The remaining agents (no + /// preference) are spread evenly over the nodes counting only themselves toward the fill, with the + /// same minimal-disruption behavior as . + /// + public void DistributeEvenlyWithAffinity(string scheme, Func preferredNodeFor) + { + if (_nodes.Count == 0) + { + throw new InvalidOperationException("There are no active nodes"); + } + + var agents = AvailableAgentsForScheme(scheme); + if (agents.Count == 0) + { + return; + } + + if (_nodes.Count == 1) + { + var only = _nodes[0]; + foreach (var agent in agents) + { + only.Assign(agent); + } + + return; + } + + var remainder = new List(); + foreach (var agent in agents) + { + var preferred = preferredNodeFor(agent.Uri); + if (preferred == null) + { + remainder.Add(agent); + continue; + } + + // Node.Assign detaches from any current node first, so an agent running away from its + // preferred node is MOVED (surfacing as a ReassignAgent command) — that one-time migration + // is what converges an existing cluster onto the per-database co-location. An agent already + // in place is left untouched, so the settled state is a fixed point. + if (!ReferenceEquals(agent.AssignedNode, preferred)) + { + preferred.Assign(agent); + } + } + + // The remainder — agents of databases with no other family's agents to follow — spreads evenly, + // counting only the remainder itself toward each node's fill. Counting the preferred placements + // too would push every no-affinity agent onto whichever nodes hold no projections, which is + // exactly backwards: those nodes have no pool open to ANY shard database yet. + var spread = (double)remainder.Count / _nodes.Count; + var minimum = (int)Math.Floor(spread); + var maximum = (int)Math.Ceiling(spread); + + foreach (var node in _nodes) + { + var extras = node.ForCurrentlyAssigned(remainder).Skip(maximum).ToArray(); + foreach (var agent in extras) + { + agent.Detach(); + } + } + + var missing = new Queue(remainder.Where(x => x.AssignedNode == null)); + + foreach (var node in _nodes) + { + if (missing.Count == 0) + { + break; + } + + var count = node.ForCurrentlyAssigned(remainder).Count(); + + for (var i = 0; i < minimum - count; i++) + { + if (missing.Count == 0) + { + break; + } + + node.Assign(missing.Dequeue()); + } + } + + while (missing.Count != 0) + { + var agent = missing.Dequeue(); + + var node = _nodes.FirstOrDefault(x => !x.IsLeader && x.ForCurrentlyAssigned(remainder).Count() < maximum) + ?? _nodes.FirstOrDefault(x => !x.IsLeader) ?? _nodes.First(); + node.Assign(agent); + } + } + public bool AllNodesHaveSameCapabilities(string scheme) { return AllNodesHaveSameCapabilities(scheme, _ => true); diff --git a/src/Wolverine/Runtime/Agents/NodeAgentController.EvaluateAssignments.cs b/src/Wolverine/Runtime/Agents/NodeAgentController.EvaluateAssignments.cs index d9429783d..a34c60842 100644 --- a/src/Wolverine/Runtime/Agents/NodeAgentController.EvaluateAssignments.cs +++ b/src/Wolverine/Runtime/Agents/NodeAgentController.EvaluateAssignments.cs @@ -1,4 +1,5 @@ using Microsoft.Extensions.Logging; +using Wolverine.Persistence; namespace Wolverine.Runtime.Agents; @@ -111,7 +112,14 @@ public async Task EvaluateAssignmentsAsync( grid.WithNode(node); } - foreach (var agentFamily in _agentFamilies.Values) + // GH-3785: the durability family places each database's agent next to that database's + // event-subscription agents, which it can only see if those were assigned first — so it must + // evaluate last. Everything else keeps its registration order (OrderBy is stable), which + // previously put the durability family last only by Dictionary insertion-order accident. + var families = _agentFamilies.Values + .OrderBy(x => x.Scheme == PersistenceConstants.AgentScheme ? 1 : 0); + + foreach (var agentFamily in families) { try {