diff --git a/docs/guide/durability/marten/distribution.md b/docs/guide/durability/marten/distribution.md index 6bc1381a5..fd6477a2c 100644 --- a/docs/guide/durability/marten/distribution.md +++ b/docs/guide/durability/marten/distribution.md @@ -59,6 +59,45 @@ Some other facts about this integration: new versions of a projection or subscription on some application nodes in order to do try ["blue/green deployment."](https://en.wikipedia.org/wiki/Blue%E2%80%93green_deployment) * This capability does depend on Wolverine's built-in [leadership election](https://en.wikipedia.org/wiki/Leader_election) -- which fortunately got a _lot_ better in Wolverine 3.0 +## Database-Affine Distribution for Sharded Stores + +By default Wolverine spreads subscription and projection agents *evenly* across the cluster (with blue/green +capability matching). That is the right choice for a store-global event store or database-per-tenant multi-tenancy, +where there is one agent per database anyway. + +It is *not* the right choice for a Marten store that combines [sharded multi-tenancy](https://martendb.io/configuration/multitenancy.html#sharded-multi-tenancy-with-database-pooling) +with [per-tenant event partitioning](https://martendb.io/events/multitenancy.html#per-tenant-event-partitioning). There, +many tenants are co-located in one shard database and each draws its own event sequence, so Wolverine fans agents out +one-per-`(shard, tenant)` rather than one-per-database. With hundreds of tenants scattered across many shard databases, +an even per-agent spread makes *every* node open a connection pool to *nearly every* shard database — so the pool count +grows as `nodes × databases` and quickly exhausts a shared server's `max_connections`. + +To bound that, opt into **database-affine assignment** on the Marten side. Wolverine then keeps every agent for a given +shard database together on a single node, so a node only opens pools to the databases it actually owns and the pool +count scales with the number of shard databases, not `nodes × databases`: + +```cs +services.AddMarten(opts => +{ + // ... sharded tenancy + per-tenant event partitioning ... + opts.Events.UseTenantPartitionedEvents = true; +}) +.IntegrateWithWolverine(x => +{ + x.UseWolverineManagedEventSubscriptionDistribution = true; + + // Keep a shard database's per-tenant agents together on one node. + x.UseDatabaseAffineAgentAssignment = true; +}); +``` + +This setting on `IntegrateWithWolverine` flows through to `StoreOptions.Events.UseDatabaseAffineAgentAssignment` +(you can also set it there directly). Wolverine reads it off the store through +`IEventStore.GroupAgentAssignmentsByDatabase`, so the behavior is entirely opt-in and store-driven — a store that +does not set it keeps the default even distribution. The grouping key is the +`[event store type]/[event store name]/[database]` prefix of the agent `Uri` (see below), so all of a shard database's +per-tenant agents share one group. + ## Uri Structure The `Uri` structure for event subscriptions or projections is: diff --git a/src/Persistence/MartenTests/Distribution/integrate_with_wolverine_affine_settings.cs b/src/Persistence/MartenTests/Distribution/integrate_with_wolverine_affine_settings.cs new file mode 100644 index 000000000..0d0763f90 --- /dev/null +++ b/src/Persistence/MartenTests/Distribution/integrate_with_wolverine_affine_settings.cs @@ -0,0 +1,63 @@ +using IntegrationTests; +using JasperFx.CodeGeneration; +using Marten; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Hosting; +using Shouldly; +using Wolverine; +using Wolverine.Marten; +using Xunit; + +namespace MartenTests.Distribution; + +/// +/// JasperFx/marten#4806: the database-affine agent-assignment setting can be configured on +/// IntegrateWithWolverine (next to UseWolverineManagedEventSubscriptionDistribution), and it must flow +/// through to the Marten event store — which is where Wolverine's distribution reads it via +/// IEventStore.GroupAgentAssignmentsByDatabase. This guards the bridge (MartenOverrides resolving the +/// MartenIntegration and applying the setting) against a DI-resolution regression that would silently +/// leave the app on the default even distribution. +/// +public class integrate_with_wolverine_affine_settings +{ + private static async Task StartHostAsync(Action configure, string schema) + { + return await Host.CreateDefaultBuilder() + .UseWolverine(opts => + { + opts.Services.AddMarten(m => + { + m.DisableNpgsqlLogging = true; + m.Connection(Servers.PostgresConnectionString); + m.DatabaseSchemaName = schema; + }) + .IntegrateWithWolverine(configure); + + opts.Discovery.DisableConventionalDiscovery(); + opts.CodeGeneration.TypeLoadMode = TypeLoadMode.Auto; + }).StartAsync(); + } + + [Fact] + public async Task affine_setting_flows_to_the_event_store() + { + using var host = await StartHostAsync(m => + { + m.UseWolverineManagedEventSubscriptionDistribution = true; + m.UseDatabaseAffineAgentAssignment = true; + }, "affine_on"); + + var store = (DocumentStore)host.Services.GetRequiredService(); + store.Options.Events.UseDatabaseAffineAgentAssignment.ShouldBeTrue(); + } + + [Fact] + public async Task default_is_unchanged_when_not_configured() + { + using var host = await StartHostAsync( + m => m.UseWolverineManagedEventSubscriptionDistribution = true, "affine_off"); + + var store = (DocumentStore)host.Services.GetRequiredService(); + store.Options.Events.UseDatabaseAffineAgentAssignment.ShouldBeFalse(); + } +} diff --git a/src/Persistence/MartenTests/Distribution/sharded_tenant_partitioned_distribution.cs b/src/Persistence/MartenTests/Distribution/sharded_tenant_partitioned_distribution.cs new file mode 100644 index 000000000..6bfdbf703 --- /dev/null +++ b/src/Persistence/MartenTests/Distribution/sharded_tenant_partitioned_distribution.cs @@ -0,0 +1,190 @@ +using IntegrationTests; +using JasperFx.CodeGeneration; +using JasperFx.Core; +using JasperFx.Events; +using JasperFx.Events.Projections; +using JasperFx.MultiTenancy; +using Marten; +using Marten.Events; +using MartenTests.Distribution.Support; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Logging; +using Npgsql; +using Shouldly; +using Weasel.Postgresql; +using Weasel.Postgresql.Migrations; +using Wolverine; +using Wolverine.Marten; +using Wolverine.Runtime.Agents; +using Wolverine.Tracking; +using Xunit; +using Xunit.Abstractions; + +namespace MartenTests.Distribution; + +// Phase 3 of the per-tenant-partitioned-events matrix — the #3280 regression itself: under +// MultiTenantedWithShardedDatabases + UseTenantPartitionedEvents, multiple tenants are CO-LOCATED in one +// shard database, each drawing its own event sequence. A single store-global agent per (shard, database) +// cannot track them (the lagging tenant's appends fall below the shared high-water mark and are skipped), +// so managed distribution must fan out one agent per (shard, tenant) — asserted here end-to-end: two +// ":all/" agents for the co-located database, and both tenants' events actually projected. +// +// NOTE: this requires the companion JasperFx.Events (jasperfx#482) + Marten (marten#4799) changes — it +// goes green in CI once those merge, per the PR's stated merge order. +public class sharded_tenant_partitioned_distribution(ITestOutputHelper output) : PostgresqlContext, IAsyncLifetime +{ + private const string TenantA = "tenant-a"; + private const string TenantB = "tenant-b"; + + private IHost theHost = null!; + private IDocumentStore theStore = null!; + + // Unique names per run: managed partitions/databases from one run otherwise break the next run's + // resource-setup DDL reconciliation. + private readonly string theSuffix = Guid.NewGuid().ToString("N")[..8]; + private string ShardDatabase => $"w3281_shard_{theSuffix}"; + private string MasterSchema => $"csp_sharded_{theSuffix}"; + + public async Task InitializeAsync() + { + // Provision the physical shard database (the master/pool lives in the default test database + // under its own schema). + await using (var conn = new NpgsqlConnection(Servers.PostgresConnectionString)) + { + await conn.OpenAsync(); + if (!await conn.DatabaseExists(ShardDatabase)) + { + await new DatabaseSpecification().BuildDatabase(conn, ShardDatabase); + } + } + + var shardConnectionString = new NpgsqlConnectionStringBuilder(Servers.PostgresConnectionString) + { + Database = ShardDatabase + }.ConnectionString; + + void ConfigureShardedStore(StoreOptions m) + { + m.DisableNpgsqlLogging = true; + m.MultiTenantedWithShardedDatabases(x => + { + x.ConnectionString = Servers.PostgresConnectionString; + x.SchemaName = MasterSchema; + x.PartitionSchemaName = MasterSchema + "_tenants"; + x.AddDatabase("shard-1", shardConnectionString); + }); + + m.Events.StreamIdentity = StreamIdentity.AsString; + m.Events.TenancyStyle = TenancyStyle.Conjoined; + m.Events.AppendMode = EventAppendMode.Quick; + m.Events.UseTenantPartitionedEvents = true; + m.Events.UseIdentityMapForAggregates = false; + + m.Schema.For().MultiTenanted(); + m.Projections.Snapshot(SnapshotLifecycle.Async); + } + + // Provision the two CO-LOCATED tenants BEFORE the Wolverine host starts, so the first agent + // enumeration already sees them and fans out per tenant. (Adding a database's FIRST tenants while + // the host runs leaves the store-global agent from the empty-database phase running alongside the + // new per-tenant agents until assignments reconcile — a transition edge outside this test's scope.) + await using (var provisioning = DocumentStore.For(ConfigureShardedStore)) + { + // Apply the event schema (parent partitioned tables) to the shard database first, so the + // per-tenant LIST partitions + sequences created by the explicit assignment have a parent. + await provisioning.Storage.ApplyAllConfiguredChangesToDatabaseAsync(); + await provisioning.Advanced.AddTenantToShardAsync(TenantA, "shard-1", default); + await provisioning.Advanced.AddTenantToShardAsync(TenantB, "shard-1", default); + } + + theHost = await Host.CreateDefaultBuilder() + .UseWolverine(opts => + { + opts.Durability.HealthCheckPollingTime = 1.Seconds(); + opts.Durability.CheckAssignmentPeriod = 1.Seconds(); + + opts.Services.AddMarten(ConfigureShardedStore) + .IntegrateWithWolverine(m => + { + m.UseWolverineManagedEventSubscriptionDistribution = true; + // Sharded tenancy has no single "main" database, so Wolverine's message store needs + // an explicit master (mirrors a real deployment pointing this at a references DB). + m.MainDatabaseConnectionString = Servers.PostgresConnectionString; + }); + + opts.Services.AddSingleton(new OutputLoggerProvider(output)); + opts.Discovery.DisableConventionalDiscovery(); + opts.CodeGeneration.TypeLoadMode = TypeLoadMode.Auto; + }).StartAsync(); + + theStore = theHost.Services.GetRequiredService(); + } + + public async Task DisposeAsync() + { + theHost.GetRuntime().Agents.DisableHealthChecks(); + await theHost.StopAsync(); + theHost.Dispose(); + } + + [Fact] + public async Task co_located_tenants_get_their_own_agents_and_both_project() + { + await theHost.WaitUntilAssumesLeadershipAsync(10.Seconds()); + + // One async projection over one shard database with two co-located tenants => one agent PER TENANT. + await theHost.WaitUntilAssignmentsChangeTo(w => + { + w.AgentScheme = EventSubscriptionAgentFamily.SchemeName; + w.ExpectRunningAgents(theHost, 2); + }, 60.Seconds()); + + var uris = await GetAgentUrisAsync(theHost); + uris.Length.ShouldBe(2); + uris.ShouldContain(u => u.TrimEnd('/').EndsWith("/" + TenantA, StringComparison.OrdinalIgnoreCase)); + uris.ShouldContain(u => u.TrimEnd('/').EndsWith("/" + TenantB, StringComparison.OrdinalIgnoreCase)); + + // Both tenants' events are projected — each tenant advances against its OWN high-water mark, so the + // second tenant's appends are not skipped below a shared mark (the #3280 failure mode). + var id = "pc-" + Guid.NewGuid().ToString("N"); + await AppendAsync(TenantA, id, 5); + await AppendAsync(TenantB, id, 11); // same stream id, other tenant partition + + (await WaitForProjectionAsync(TenantA, id, 60.Seconds()))!.Total.ShouldBe(5); + (await WaitForProjectionAsync(TenantB, id, 60.Seconds()))!.Total.ShouldBe(11); + } + + private async Task AppendAsync(string tenant, string id, int amount) + { + await using var session = theStore.LightweightSession(tenant); + session.Events.StartStream(id, new CounterChanged(amount)); + await session.SaveChangesAsync(); + } + + private async Task WaitForProjectionAsync(string tenant, string id, TimeSpan timeout) + { + var deadline = DateTimeOffset.UtcNow + timeout; + while (DateTimeOffset.UtcNow < deadline) + { + await using var session = theStore.QuerySession(tenant); + var doc = await session.LoadAsync(id); + if (doc != null) + { + return doc; + } + + await Task.Delay(250.Milliseconds()); + } + + return null; + } + + private static async Task GetAgentUrisAsync(IHost host) + { + var family = host.Services.GetServices() + .OfType().Single(); + var agents = await family.AllKnownAgentsAsync(); + return [.. agents.Select(x => x.AbsoluteUri)]; + } +} diff --git a/src/Persistence/Wolverine.Marten/MartenIntegration.cs b/src/Persistence/Wolverine.Marten/MartenIntegration.cs index abceea335..852cbd2fc 100644 --- a/src/Persistence/Wolverine.Marten/MartenIntegration.cs +++ b/src/Persistence/Wolverine.Marten/MartenIntegration.cs @@ -40,12 +40,24 @@ public class MartenIntegration : IWolverineExtension, IEventForwarding public bool UseFastEventForwarding { get; set; } /// - /// Use this when using Wolverine to evenly distribute event projection and subscription - /// work of Marten asynchronous projections. This replaces Marten's AddAsyncDaemon(HotCold) + /// Use this when using Wolverine to evenly distribute event projection and subscription + /// work of Marten asynchronous projections. This replaces Marten's AddAsyncDaemon(HotCold) /// option and should not be used in combination with Marten's own load distribution. /// public bool UseWolverineManagedEventSubscriptionDistribution { get; set; } + /// + /// Opt-in (default false): when is on and + /// the Marten store uses sharded databases with per-tenant event partitioning, assign each shard + /// database's per-(shard, tenant) agents with database affinity — all of a database's agents run + /// on one node — instead of spreading them evenly by count. This bounds each node to the shard databases + /// it owns, so connection pools scale with the number of shard databases rather than nodes × databases + /// (which otherwise exhausts a shared server's max_connections). Flows through to + /// StoreOptions.Events.UseDatabaseAffineAgentAssignment; only takes effect for a sharded, + /// per-tenant-partitioned store. See JasperFx/marten#4806. + /// + public bool UseDatabaseAffineAgentAssignment { get; set; } + public void Configure(WolverineOptions options) { // Duplicate incoming messages @@ -201,6 +213,17 @@ public void Configure(IServiceProvider services, StoreOptions options) nameof(Saga.Version), typeof(int))!; } }); + + // Bridge the Wolverine-managed distribution's database-affine agent-assignment setting — configured + // on IntegrateWithWolverine — onto the Marten event store, because Wolverine's distribution is what + // consumes it (via IEventStore.GroupAgentAssignmentsByDatabase). Only the main store + // (StoreType == null): this is a sharded per-tenant-store concern, not ancillary. + // See JasperFx/marten#4806. + if (StoreType == null && + services.GetService() is { UseDatabaseAffineAgentAssignment: true }) + { + options.Events.UseDatabaseAffineAgentAssignment = true; + } } } diff --git a/src/Testing/CoreTests/Runtime/Agents/distribute_by_group_affinity.cs b/src/Testing/CoreTests/Runtime/Agents/distribute_by_group_affinity.cs new file mode 100644 index 000000000..335f0b311 --- /dev/null +++ b/src/Testing/CoreTests/Runtime/Agents/distribute_by_group_affinity.cs @@ -0,0 +1,96 @@ +using Wolverine.Runtime.Agents; +using Xunit; + +namespace CoreTests.Runtime.Agents; + +// JasperFx/marten#4806: AssignmentGrid.DistributeByGroupAffinity keeps every agent of a group (e.g. one +// shard database) on the same node, and spreads whole groups across nodes — so a node opens connection +// pools only to the databases it owns, instead of every node touching every shard database. +public class distribute_by_group_affinity +{ + // Group key = the database segment of an event-subscriptions agent URI: + // event-subscriptions://{type}/{name}/{databaseId}/{shard...} -> Segments[2]. + private static string DatabaseKey(Uri uri) => uri.Segments[2].Trim('/'); + + private static Uri Agent(string db, string tenant) => + new($"event-subscriptions://marten/main/{db}/Proj:All:{tenant}"); + + [Fact] + public void keeps_a_databases_agents_together_and_spreads_databases_across_nodes() + { + var grid = new AssignmentGrid(); + grid.WithNode(1, Guid.NewGuid()); + grid.WithNode(2, Guid.NewGuid()); + + // 4 shard databases, 3 tenants each = 12 per-tenant agents. + var agents = new List(); + foreach (var db in new[] { "db1", "db2", "db3", "db4" }) + foreach (var tenant in new[] { "t1", "t2", "t3" }) + agents.Add(Agent(db, tenant)); + + grid.WithAgents(agents.ToArray()); + + grid.DistributeByGroupAffinity("event-subscriptions", DatabaseKey); + + // Every tenant agent of a database lands on exactly one node (the database is never split). + foreach (var db in new[] { "db1", "db2", "db3", "db4" }) + { + var nodes = new[] { "t1", "t2", "t3" } + .Select(t => grid.AgentFor(Agent(db, t)).AssignedNode) + .Distinct() + .ToList(); + + nodes.Count.ShouldBe(1, $"all agents of {db} must be on a single node"); + nodes[0].ShouldNotBeNull(); + } + + // All agents are assigned, and the 4 databases are spread across both nodes (2 each). + grid.AllAgents.ShouldAllBe(a => a.AssignedNode != null); + var perNode = grid.Nodes.Select(n => n.ForScheme("event-subscriptions").Count()).OrderBy(x => x).ToList(); + perNode.ShouldBe(new[] { 6, 6 }); + } + + [Fact] + public void single_node_takes_every_agent() + { + var grid = new AssignmentGrid(); + var node = grid.WithNode(1, Guid.NewGuid()); + + grid.WithAgents(Agent("db1", "t1"), Agent("db1", "t2"), Agent("db2", "t1")); + + grid.DistributeByGroupAffinity("event-subscriptions", DatabaseKey); + + node.ForScheme("event-subscriptions").Count().ShouldBe(3); + } + + [Fact] + public void a_heavy_database_stays_on_one_node_and_light_groups_balance_around_it() + { + var grid = new AssignmentGrid(); + grid.WithNode(1, Guid.NewGuid()); + grid.WithNode(2, Guid.NewGuid()); + + // One heavy database (4 tenants) + two light ones (1 tenant each). Largest-first placement puts the + // heavy group on one node and both light groups on the other — totals stay as balanced as whole + // groups allow, and a database is never split across nodes. + var agents = new List(); + foreach (var tenant in new[] { "t1", "t2", "t3", "t4" }) + agents.Add(Agent("dbHeavy", tenant)); + agents.Add(Agent("dbLight1", "t1")); + agents.Add(Agent("dbLight2", "t1")); + + grid.WithAgents(agents.ToArray()); + + grid.DistributeByGroupAffinity("event-subscriptions", DatabaseKey); + + grid.AllAgents.ShouldAllBe(a => a.AssignedNode != null); + + var heavyNode = grid.AgentFor(Agent("dbHeavy", "t1")).AssignedNode; + new[] { "t2", "t3", "t4" } + .Select(t => grid.AgentFor(Agent("dbHeavy", t)).AssignedNode) + .ShouldAllBe(n => n == heavyNode); + + grid.AgentFor(Agent("dbLight1", "t1")).AssignedNode.ShouldNotBe(heavyNode); + grid.AgentFor(Agent("dbLight2", "t1")).AssignedNode.ShouldNotBe(heavyNode); + } +} diff --git a/src/Testing/CoreTests/Runtime/Agents/event_store_agents_per_tenant_distribution.cs b/src/Testing/CoreTests/Runtime/Agents/event_store_agents_per_tenant_distribution.cs new file mode 100644 index 000000000..351e451d6 --- /dev/null +++ b/src/Testing/CoreTests/Runtime/Agents/event_store_agents_per_tenant_distribution.cs @@ -0,0 +1,111 @@ +using System; +using System.Linq; +using System.Threading; +using System.Threading.Tasks; +using JasperFx.Descriptors; +using JasperFx.Events; +using JasperFx.Events.Daemon; +using JasperFx.Events.Descriptors; +using JasperFx.Events.Projections; +using NSubstitute; +using Shouldly; +using Wolverine.Runtime.Agents; +using Xunit; + +namespace CoreTests.Runtime.Agents; + +/// +/// wolverine#3280: under sharded databases + per-tenant event partitioning multiple tenants are +/// co-located in one shard database, each drawing its own event sequence, so a single store-global agent +/// per (shard, database) cannot track them. When the store reports , +/// must fan out one agent per (shard, tenant); for any +/// other store it must keep the single store-global agent. +/// +public class event_store_agents_per_tenant_distribution +{ + private static EventStoreUsage UsageWithOneAsyncShardOnADatabaseWith(params string[] tenantIds) + { + return new EventStoreUsage + { + Database = new DatabaseUsage + { + Databases = + { + new DatabaseDescriptor + { + ServerName = "localhost", + DatabaseName = "claims1", + TenantIds = tenantIds.ToList() + } + } + }, + Subscriptions = + { + new SubscriptionDescriptor(SubscriptionType.SingleStreamProjection) + { + Lifecycle = ProjectionLifecycle.Async, + ShardNames = new[] { ShardName.Compose("provided-cares") } + } + } + }; + } + + private static EventStoreAgents AgentsFor(EventStoreUsage usage, bool distributesPerTenant) + { + var store = Substitute.For(); + store.Identity.Returns(new EventStoreIdentity("main", "marten")); + store.DistributesAgentsPerTenant.Returns(distributesPerTenant); + store.TryCreateUsage(Arg.Any()).Returns(Task.FromResult(usage)); + return new EventStoreAgents(store, Array.Empty>()); + } + + [Fact] + public async Task fans_out_one_agent_per_tenant_when_the_store_distributes_per_tenant() + { + var agents = AgentsFor(UsageWithOneAsyncShardOnADatabaseWith("01059910", "08004464"), distributesPerTenant: true); + + var uris = await agents.SupportedAgentsAsync(CancellationToken.None); + + uris.Count.ShouldBe(2, "one async shard on a two-tenant database must yield one agent per tenant"); + uris.ShouldContain(u => u.AbsolutePath.TrimEnd('/').EndsWith("/01059910", StringComparison.OrdinalIgnoreCase)); + uris.ShouldContain(u => u.AbsolutePath.TrimEnd('/').EndsWith("/08004464", StringComparison.OrdinalIgnoreCase)); + } + + [Fact] + public async Task keeps_a_single_store_global_agent_when_the_store_does_not_distribute_per_tenant() + { + var agents = AgentsFor(UsageWithOneAsyncShardOnADatabaseWith("01059910", "08004464"), distributesPerTenant: false); + + var uris = await agents.SupportedAgentsAsync(CancellationToken.None); + + uris.Count.ShouldBe(1, "without per-tenant distribution a shard keeps its single store-global agent"); + uris.ShouldNotContain(u => u.AbsolutePath.TrimEnd('/').EndsWith("/01059910", StringComparison.OrdinalIgnoreCase)); + } + + [Fact] + public async Task keeps_a_single_agent_when_the_database_has_no_tenants_yet() + { + var agents = AgentsFor(UsageWithOneAsyncShardOnADatabaseWith(), distributesPerTenant: true); + + var uris = await agents.SupportedAgentsAsync(CancellationToken.None); + + uris.Count.ShouldBe(1, "a per-tenant store with no tenants provisioned yet keeps one store-global agent"); + } + + [Fact] + public async Task dedupes_tenants_case_insensitively_and_skips_blank_ids() + { + // Tenant ids are matched case-insensitively elsewhere (TryResolveTenantDatabaseIdAsync), so the + // fan-out must not produce two agents for the same logical tenant in different casing, nor an + // invalid agent for a blank id that slipped into the usage descriptor. + var agents = AgentsFor( + UsageWithOneAsyncShardOnADatabaseWith("Clinic-A", "clinic-a", "clinic-b", "", " "), + distributesPerTenant: true); + + var uris = await agents.SupportedAgentsAsync(CancellationToken.None); + + uris.Count.ShouldBe(2, "one agent per logical tenant: clinic-a (deduped across casing) and clinic-b"); + uris.ShouldContain(u => u.AbsolutePath.TrimEnd('/').EndsWith("/clinic-a", StringComparison.OrdinalIgnoreCase)); + uris.ShouldContain(u => u.AbsolutePath.TrimEnd('/').EndsWith("/clinic-b", StringComparison.OrdinalIgnoreCase)); + } +} diff --git a/src/Testing/CoreTests/Runtime/Agents/event_subscription_family_affine_assignment.cs b/src/Testing/CoreTests/Runtime/Agents/event_subscription_family_affine_assignment.cs new file mode 100644 index 000000000..2b52fbfef --- /dev/null +++ b/src/Testing/CoreTests/Runtime/Agents/event_subscription_family_affine_assignment.cs @@ -0,0 +1,124 @@ +using System; +using System.Collections.Generic; +using System.Linq; +using System.Threading; +using System.Threading.Tasks; +using JasperFx.Events; +using NSubstitute; +using Shouldly; +using Wolverine.Runtime.Agents; +using Xunit; + +namespace CoreTests.Runtime.Agents; + +/// +/// JasperFx/marten#4806: when a sharded store opts into , +/// must assign that store's per-(shard, tenant) +/// agents with database affinity (a database's agents kept together on one node) instead of the default even +/// blue/green spread. These are fast unit tests: agents are registered directly on an +/// and the store flag is stubbed, so no host or database is required. +/// +public class event_subscription_family_affine_assignment +{ + // Agent URI grammar is event-subscriptions://{type}/{name}/{databaseId}/{shard...} (see EventStoreAgents.UriFor). + private static Uri Agent(string db, string tenant) => + new($"event-subscriptions://marten/main/{db}/Proj:All:{tenant}"); + + private static EventSubscriptionAgentFamily FamilyFor(bool affine) + { + var store = Substitute.For(); + store.Identity.Returns(new EventStoreIdentity("main", "marten")); + store.GroupAgentAssignmentsByDatabase.Returns(affine); + return new EventSubscriptionAgentFamily(new[] { store }, Array.Empty>()); + } + + [Fact] + public async Task keeps_a_databases_agents_together_when_the_store_opts_into_affinity() + { + var family = FamilyFor(affine: true); + + var grid = new AssignmentGrid(); + grid.WithNode(1, Guid.NewGuid()); + grid.WithNode(2, Guid.NewGuid()); + // One shard database with two tenants: even distribution would split it across the two nodes. + grid.WithAgents(Agent("dbShared", "t1"), Agent("dbShared", "t2")); + + await family.EvaluateAssignmentsAsync(grid); + + var nodes = new[] { "t1", "t2" } + .Select(t => grid.AgentFor(Agent("dbShared", t)).AssignedNode) + .Distinct() + .ToList(); + + nodes.Count.ShouldBe(1, "an affine store must keep a shard database's agents on a single node"); + nodes[0].ShouldNotBeNull(); + } + + [Fact] + public async Task spreads_a_databases_agents_evenly_when_the_store_does_not_opt_in() + { + var family = FamilyFor(affine: false); + + var grid = new AssignmentGrid(); + grid.WithNode(1, Guid.NewGuid()); + grid.WithNode(2, Guid.NewGuid()); + grid.WithAgents(Agent("dbShared", "t1"), Agent("dbShared", "t2")); + + await family.EvaluateAssignmentsAsync(grid); + + var nodes = new[] { "t1", "t2" } + .Select(t => grid.AgentFor(Agent("dbShared", t)).AssignedNode) + .Distinct() + .ToList(); + + nodes.Count.ShouldBe(2, "without affinity even distribution splits a two-agent database across both nodes"); + } + + [Fact] + public void database_key_groups_a_databases_agents_and_separates_databases() + { + var t1 = EventSubscriptionAgentFamily.DatabaseKeyOf(Agent("claims1", "t1")); + var t2 = EventSubscriptionAgentFamily.DatabaseKeyOf(Agent("claims1", "t2")); + var other = EventSubscriptionAgentFamily.DatabaseKeyOf(Agent("claims2", "t1")); + + t1.ShouldBe("marten/main/claims1"); + t2.ShouldBe(t1, "two tenants on the same shard database must share a group key"); + other.ShouldNotBe(t1, "a different shard database must be a different group"); + } + + [Fact] + public void database_key_falls_back_to_the_whole_uri_when_there_is_no_database_segment() + { + // Defensive: a URI without the {databaseId} segment (fewer than 3 path segments) can't be grouped by + // database, so each such agent is its own group rather than silently collapsing together. + var uri = new Uri("event-subscriptions://marten/main"); + EventSubscriptionAgentFamily.DatabaseKeyOf(uri).ShouldBe(uri.AbsoluteUri); + } +} + +/// +/// surfaces the affine-assignment flag from its underlying +/// unchanged; the agent family reads it off the wrapper. See JasperFx/marten#4806. +/// +public class event_store_agents_affinity_passthrough +{ + private static EventStoreAgents AgentsFor(bool affine) + { + var store = Substitute.For(); + store.Identity.Returns(new EventStoreIdentity("main", "marten")); + store.GroupAgentAssignmentsByDatabase.Returns(affine); + return new EventStoreAgents(store, Array.Empty>()); + } + + [Fact] + public void surfaces_group_by_database_from_the_store() + { + AgentsFor(affine: true).GroupAgentAssignmentsByDatabase.ShouldBeTrue(); + } + + [Fact] + public void default_flows_through_for_a_non_affine_store() + { + AgentsFor(affine: false).GroupAgentAssignmentsByDatabase.ShouldBeFalse(); + } +} diff --git a/src/Wolverine/Runtime/Agents/AssignmentGrid.Distribution.cs b/src/Wolverine/Runtime/Agents/AssignmentGrid.Distribution.cs index 6489a50cd..963847e48 100644 --- a/src/Wolverine/Runtime/Agents/AssignmentGrid.Distribution.cs +++ b/src/Wolverine/Runtime/Agents/AssignmentGrid.Distribution.cs @@ -83,6 +83,74 @@ public void DistributeEvenly(string scheme) } } + /// + /// Distribute agents of a scheme across nodes with group affinity: all agents that share a + /// (e.g. a shard database) are assigned to the same node. Intended for + /// sharded event stores whose per-(shard, tenant) agents each connect to their shard database: an even + /// per-agent spread makes every node open pools to (nearly) every database (pools grow as + /// nodes×databases and exhaust a shared server's max_connections), while grouping keeps each node + /// connected only to the databases it owns, so pools scale with the number of databases + /// (JasperFx/marten#4806). + /// + /// Groups are placed largest-first onto the least-loaded node (deterministic tie-breaks), so total + /// agent count stays balanced and a steady grid does not churn. This does not apply blue/green + /// capability matching; use when that is + /// required. + /// + public void DistributeByGroupAffinity(string scheme, Func groupKey) + { + if (_nodes.Count == 0) + { + throw new InvalidOperationException("There are no active nodes"); + } + + var agents = AvailableAgentsForScheme(scheme); + if (agents.Count == 0) + { + return; + } + + var nodes = _nodes.OrderBy(x => x.IsLeader).ThenBy(x => x.AssignedId).ToList(); + + if (nodes.Count == 1) + { + var only = nodes[0]; + foreach (var agent in agents) + { + only.Assign(agent); + } + + return; + } + + var load = nodes.ToDictionary(n => n, _ => 0); + + var groups = agents + .GroupBy(a => groupKey(a.Uri)) + .OrderByDescending(g => g.Count()) + .ThenBy(g => g.Key, StringComparer.Ordinal) + .ToList(); + + foreach (var group in groups) + { + var members = group.ToList(); + + // The least-loaded node hosts the whole group (tie-breaks: non-leader first, then node id). + var node = load + .OrderBy(kv => kv.Value) + .ThenBy(kv => kv.Key.IsLeader) + .ThenBy(kv => kv.Key.AssignedId) + .First().Key; + + foreach (var agent in members) + { + node.Assign(agent); + } + + load[node] += members.Count; + } + } + public bool AllNodesHaveSameCapabilities(string scheme) { var gold = _nodes[0].OrderedCapabilitiesForScheme(scheme); diff --git a/src/Wolverine/Runtime/Agents/EventStoreAgents.cs b/src/Wolverine/Runtime/Agents/EventStoreAgents.cs index 1767cd722..5adacf91e 100644 --- a/src/Wolverine/Runtime/Agents/EventStoreAgents.cs +++ b/src/Wolverine/Runtime/Agents/EventStoreAgents.cs @@ -30,6 +30,14 @@ public EventStoreAgents(IEventStore store, IObserver[] observers) public EventStoreIdentity Identity { get; } + /// + /// When true (opt-in, sharded stores only), this store's per-(shard, tenant) agents should be assigned + /// with database affinity — all agents for a shard database on one node — so a node opens pools only to + /// the databases it owns. Surfaces . See + /// JasperFx/marten#4806. + /// + public bool GroupAgentAssignmentsByDatabase => _store.GroupAgentAssignmentsByDatabase; + public async ValueTask DisposeAsync() { foreach (var entry in _daemons.Enumerate()) @@ -87,18 +95,48 @@ public async ValueTask> SupportedAgentsAsync(CancellationToke var usage = await _store.TryCreateUsage(cancellation); if (usage == null) return list; + var asyncShards = usage.Subscriptions + .Where(x => x.Lifecycle == ProjectionLifecycle.Async) + .SelectMany(x => x.ShardNames) + .ToArray(); + + // wolverine#3280: under sharded databases + per-tenant event partitioning, multiple tenants are + // co-located in one shard database and each draws its own event sequence, so a single store-global + // agent per (shard, database) cannot track them — fan out one agent per (shard, tenant) instead. + // For any other store (or a database with no tenants yet) keep the one store-global agent. + void addAgentsForShard(DatabaseDescriptor database, DatabaseId id, ShardName shardName) + { + if (_store.DistributesAgentsPerTenant && database.TenantIds.Count > 0) + { + // Tenant ids are matched case-insensitively elsewhere in this class (see + // TryResolveTenantDatabaseIdAsync), so dedupe the fan-out the same way, and skip blank ids — + // a malformed usage descriptor must not yield duplicate or invalid per-tenant agent URIs. + foreach (var tenantId in database.TenantIds + .Where(t => !string.IsNullOrWhiteSpace(t)) + .Distinct(StringComparer.OrdinalIgnoreCase)) + { + var tenantShard = shardName.ForTenant(tenantId); + _shardNames = _shardNames.AddOrUpdate(tenantShard.RelativeUrl, tenantShard); + list.Add(EventSubscriptionAgentFamily.UriFor(_store.Identity, id, tenantShard)); + } + } + else + { + _shardNames = _shardNames.AddOrUpdate(shardName.RelativeUrl, shardName); + list.Add(EventSubscriptionAgentFamily.UriFor(_store.Identity, id, shardName)); + } + } + // Using this to keep from double dipping var databaseIds = new List(); foreach (var database in usage.Database.Databases) { var id = new DatabaseId(database.ServerName, database.DatabaseName); databaseIds.Add(id); - - foreach (var shardName in usage.Subscriptions.Where(x => x.Lifecycle == ProjectionLifecycle.Async).SelectMany(x => x.ShardNames)) + + foreach (var shardName in asyncShards) { - _shardNames = _shardNames.AddOrUpdate(shardName.RelativeUrl, shardName); - var uri = EventSubscriptionAgentFamily.UriFor(_store.Identity, id, shardName); - list.Add(uri); + addAgentsForShard(database, id, shardName); } } @@ -108,11 +146,9 @@ public async ValueTask> SupportedAgentsAsync(CancellationToke var id = new DatabaseId(database.ServerName, database.DatabaseName); if (!databaseIds.Contains(id)) { - foreach (var shardName in usage.Subscriptions.Where(x => x.Lifecycle == ProjectionLifecycle.Async).SelectMany(x => x.ShardNames)) + foreach (var shardName in asyncShards) { - _shardNames = _shardNames.AddOrUpdate(shardName.RelativeUrl, shardName); - var uri = EventSubscriptionAgentFamily.UriFor(_store.Identity, id, shardName); - list.Add(uri); + addAgentsForShard(database, id, shardName); } } } diff --git a/src/Wolverine/Runtime/Agents/EventSubscriptionAgentFamily.cs b/src/Wolverine/Runtime/Agents/EventSubscriptionAgentFamily.cs index 74b564a60..dd8279f8d 100644 --- a/src/Wolverine/Runtime/Agents/EventSubscriptionAgentFamily.cs +++ b/src/Wolverine/Runtime/Agents/EventSubscriptionAgentFamily.cs @@ -156,10 +156,34 @@ public async ValueTask> SupportedAgentsAsync() public ValueTask EvaluateAssignmentsAsync(AssignmentGrid assignments) { - assignments.DistributeEvenlyWithBlueGreenSemantics(SchemeName); + // JasperFx/marten#4806 (opt-in): when a sharded store asks for database-affine assignment, keep a + // shard database's per-tenant agents together on one node so a node opens connection pools only to + // the databases it owns (pools scale with databases, not nodes × databases, which otherwise + // exhausts a shared server). Otherwise distribute evenly with blue/green semantics as before. + var anyAffineStore = _stores.Enumerate() + .Any(e => e.Value.GroupAgentAssignmentsByDatabase); + + if (anyAffineStore) + { + // One grid pass covers every store's agents — their per-store URI prefix keeps groups distinct. + assignments.DistributeByGroupAffinity(SchemeName, DatabaseKeyOf); + } + else + { + assignments.DistributeEvenlyWithBlueGreenSemantics(SchemeName); + } + return new ValueTask(); } + // Agent URIs are event-subscriptions://{type}/{name}/{databaseId}/{shard...} (see UriFor); the + // (type, name, databaseId) prefix identifies the shard database an agent belongs to, so grouping on it + // co-locates every tenant/projection agent for one database on the same node. + internal static string DatabaseKeyOf(Uri uri) + => uri.Segments.Length >= 3 + ? $"{uri.Host}/{uri.Segments[1].Trim('/')}/{uri.Segments[2].Trim('/')}" + : uri.AbsoluteUri; + public async ValueTask DisposeAsync() { foreach (var store in _stores.Enumerate())