Per-tenant agent distribution + database-affine assignment surface (wolverine#3280, marten#4806)

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<EventStoreUsage?> TryCreateUsage(CancellationToken token) => throw new NotImplementedException();
+        public Uri Subject => throw new NotImplementedException();
+
+        public ValueTask<IProjectionDaemon> BuildProjectionDaemonAsync(
+            string? tenantIdOrDatabaseIdentifier = null, ILogger? logger = null)
+            => throw new NotImplementedException();
+
+        public ValueTask<IProjectionDaemon> 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();
+    }
+}

src/JasperFx.Events/Daemon/JasperFxAsyncDaemon.cs

@@ -227,6 +227,46 @@ public async Task StartAgentAsync(string shardName, CancellationToken token)
         }
 
 
+        // wolverine#3280: a per-tenant identity ("<proj>:All:<tenant>", or versioned
+        // "<proj>:V{n}:All:<tenant>") 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 <proj>:All:<tenant> 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)
         {

src/JasperFx.Events/IEventStore.cs

@@ -37,6 +37,32 @@ ValueTask<IProjectionDaemon> BuildProjectionDaemonAsync(
     DatabaseCardinality DatabaseCardinality { get; }
     bool HasMultipleTenants { get; }
 
+    /// <summary>
+    ///     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.
+    /// </summary>
+    bool DistributesAgentsPerTenant => false;
+
+    /// <summary>
+    ///     When true, a node-distributed async daemon should assign the per-(shard, tenant) agents with
+    ///     <b>database affinity</b> — every agent for the same shard database is placed on the same node —
+    ///     rather than spreading individual agents evenly by count. This complements
+    ///     <see cref="DistributesAgentsPerTenant" /> (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.
+    /// </summary>
+    bool GroupAgentAssignmentsByDatabase => false;
+
     /// <summary>
     ///     Resolve every <see cref="IEventDatabase" /> backing this event store, store-agnostically.
     ///     This is the store-neutral counterpart to Marten's <c>IMartenStorage.AllDatabases()</c> — it lets