diff --git a/src/EventTests/AgentDistributionDefaultTests.cs b/src/EventTests/AgentDistributionDefaultTests.cs new file mode 100644 index 00000000..fac66858 --- /dev/null +++ b/src/EventTests/AgentDistributionDefaultTests.cs @@ -0,0 +1,62 @@ +using System; +using System.Diagnostics; +using System.Diagnostics.Metrics; +using System.Threading; +using System.Threading.Tasks; +using JasperFx.Descriptors; +using JasperFx.Events; +using JasperFx.Events.Daemon; +using JasperFx.Events.Descriptors; +using Microsoft.Extensions.Logging; +using Shouldly; + +namespace EventTests; + +public class AgentDistributionDefaultTests +{ + // A bare IEventStore that overrides none of the agent-distribution members, so the assertions below + // exercise the default interface implementations. A store that does not opt in must behave exactly as + // before: one agent per shard×database (DistributesAgentsPerTenant false) and even per-agent spreading + // (GroupAgentAssignmentsByDatabase false). See jasperfx/wolverine#3280 and JasperFx/marten#4806. + // Everything else throws — those members are never invoked here. + private sealed class BareEventStore : IEventStore + { + public Task TryCreateUsage(CancellationToken token) => throw new NotImplementedException(); + public Uri Subject => throw new NotImplementedException(); + + public ValueTask BuildProjectionDaemonAsync( + string? tenantIdOrDatabaseIdentifier = null, ILogger? logger = null) + => throw new NotImplementedException(); + + public ValueTask BuildProjectionDaemonAsync(DatabaseId id) + => throw new NotImplementedException(); + + public Meter Meter => throw new NotImplementedException(); + public ActivitySource ActivitySource => throw new NotImplementedException(); + public string MetricsPrefix => throw new NotImplementedException(); + public DatabaseCardinality DatabaseCardinality => throw new NotImplementedException(); + public bool HasMultipleTenants => throw new NotImplementedException(); + public EventStoreIdentity Identity => throw new NotImplementedException(); + public IReadOnlyEventStore OpenReadOnlyEventStore() => throw new NotImplementedException(); + + public Task CompactStreamAsync(Guid streamId, CancellationToken token = default) + => throw new NotImplementedException(); + + public Task CompactStreamAsync(string streamKey, CancellationToken token = default) + => throw new NotImplementedException(); + } + + private readonly IEventStore theStore = new BareEventStore(); + + [Fact] + public void distributes_agents_per_tenant_defaults_to_false() + { + theStore.DistributesAgentsPerTenant.ShouldBeFalse(); + } + + [Fact] + public void group_agent_assignments_by_database_defaults_to_false() + { + theStore.GroupAgentAssignmentsByDatabase.ShouldBeFalse(); + } +} diff --git a/src/JasperFx.Events/Daemon/JasperFxAsyncDaemon.cs b/src/JasperFx.Events/Daemon/JasperFxAsyncDaemon.cs index 7beeec8d..513fadc6 100644 --- a/src/JasperFx.Events/Daemon/JasperFxAsyncDaemon.cs +++ b/src/JasperFx.Events/Daemon/JasperFxAsyncDaemon.cs @@ -227,6 +227,46 @@ public async Task StartAgentAsync(string shardName, CancellationToken token) } + // wolverine#3280: a per-tenant identity (":All:", or versioned + // ":V{n}:All:") is requested individually under node-distributed daemons + // (Wolverine-managed distribution). AllShards() only carries the store-global identities, so + // resolve the BASE shard and fan out a per-tenant agent — the same shape + // buildPerTenantContinuousAgents uses — activating the tenant in the high-water coordinator so it + // advances against its own mark and persists its own :All: progression row. + if (ShardName.TryParse(shardName, out var requested) && requested?.TenantId != null) + { + var baseIdentity = ShardName.Compose(requested.Name, requested.ShardKey, null, requested.Version).Identity; + var baseShard = _store.AllShards().FirstOrDefault(x => x.Name.Identity == baseIdentity); + if (baseShard == null) + { + throw new ArgumentOutOfRangeException(nameof(shardName), + $"Unknown shard name '{shardName}'. Value options are {_store.AllShards().Select(x => x.Name.Identity).Join(", ")}"); + } + + _tenantHighWater?.PolledTenants.Activate(requested.TenantId); + var tenantShard = baseShard with { Name = baseShard.Name.ForTenant(requested.TenantId) }; + var tenantAgent = buildAgentForShard(tenantShard); + var tenantStarted = await tryStartAgentAsync(tenantAgent, ShardExecutionMode.Continuous).ConfigureAwait(false); + if (!tenantStarted && tenantAgent is IAsyncDisposable td) + { + await td.DisposeAsync().ConfigureAwait(false); + } + else if (tenantStarted) + { + // wolverine#3280: unlike StartAllAsync (which primes every tenant's ceiling before + // starting its agents), Wolverine-managed distribution starts each tenant agent + // individually here — so its ceiling is not yet known and tryStartAgentAsync seeds it at + // high-water 0. The per-tenant poll only re-runs on a *global* high-water tick, which the + // store broadcasts only when the store-global mark CHANGES (HighWaterAgent publishes only on + // change); once it is stable (e.g. a node that starts after catch-up), no tick follows and + // the freshly-started agent would sit idle at 0 forever. Drive one poll now so it is routed + // its own tenant's mark and advances. + await pollTenantHighWaterAsync().ConfigureAwait(false); + } + + return; + } + var shard = _store.AllShards().FirstOrDefault(x => x.Name.Identity == shardName); if (shard == null) { diff --git a/src/JasperFx.Events/IEventStore.cs b/src/JasperFx.Events/IEventStore.cs index 5a72fb0f..079c9a4f 100644 --- a/src/JasperFx.Events/IEventStore.cs +++ b/src/JasperFx.Events/IEventStore.cs @@ -37,6 +37,32 @@ ValueTask BuildProjectionDaemonAsync( DatabaseCardinality DatabaseCardinality { get; } bool HasMultipleTenants { get; } + /// + /// When true, a node-distributed async daemon (e.g. Wolverine-managed event-subscription + /// distribution) must fan out one subscription agent per (shard, tenant) rather than a single + /// store-global agent per (shard, database). Required for stores where multiple tenants are + /// co-located in one database and each tenant draws its own event sequence (sharded databases + + /// per-tenant event partitioning): a single store-global high-water cannot track tenants whose + /// independent sequences overlap, so a lagging tenant's later appends would fall below it and be + /// skipped. Default false (one agent per shard×database) — correct for store-global, single-database + /// per-tenant partitioning, and database-per-tenant stores. See jasperfx/wolverine#3280. + /// + bool DistributesAgentsPerTenant => false; + + /// + /// When true, a node-distributed async daemon should assign the per-(shard, tenant) agents with + /// database affinity — every agent for the same shard database is placed on the same node — + /// rather than spreading individual agents evenly by count. This complements + /// (which fans agents out per tenant): with many 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 + /// exhausts a shared server's max_connections. Grouping a database's agents onto one node bounds + /// each node to the databases it owns, so pools scale with databases, not nodes×databases. Default + /// false (even per-agent distribution). Only meaningful for sharded-database stores. + /// See JasperFx/marten#4806. + /// + bool GroupAgentAssignmentsByDatabase => false; + /// /// Resolve every backing this event store, store-agnostically. /// This is the store-neutral counterpart to Marten's IMartenStorage.AllDatabases() — it lets