Managed distribution: per-tenant agents + database-affine assignment for sharded stores (#3280, marten#4806)

src/Wolverine/Runtime/Agents/EventStoreAgents.cs

@@ -87,18 +87,43 @@
         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)
+            {
+                foreach (var tenantId in database.TenantIds.Distinct())
+                {
+                    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<DatabaseId>();
         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 +133,9 @@
             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);
                 }
             }
         }