diff --git a/src/Persistence/MartenTests/Distribution/sharded_two_databases_affine_colocation.cs b/src/Persistence/MartenTests/Distribution/sharded_two_databases_affine_colocation.cs index c1460cc74..67dbe0dad 100644 --- a/src/Persistence/MartenTests/Distribution/sharded_two_databases_affine_colocation.cs +++ b/src/Persistence/MartenTests/Distribution/sharded_two_databases_affine_colocation.cs @@ -162,5 +162,27 @@ await leader.WaitUntilAssignmentsChangeTo(w => // And the two databases are actually spread — one per node, not both piled on one. byDatabase.Select(g => g.Select(x => x.Node).Distinct().Single()).Distinct().Count().ShouldBe(2); + + // GH-3340: each node exposes the databases it owns via the public API, matching exactly the + // databases of the agents actually running on it — so per-node work doesn't fan out to every shard. + foreach (var host in _hosts) + { + var expected = host.GetRuntime().Agents.AllRunningAgentUris() + .Where(u => u.Scheme == EventSubscriptionAgentFamily.SchemeName) + .Select(EventSubscriptionAgentFamily.DatabaseIdOf) + .Where(id => id != null) + .Select(id => id!) + .Distinct() + .ToList(); + + var owned = host.GetRuntime().Agents.AllLocallyOwnedDatabaseIds(); + + owned.Count.ShouldBe(1, "each node owns exactly one of the two shard databases"); + owned.ShouldBe(expected); + } + + // The two nodes own disjoint database sets that together cover both shard databases. + _hosts.SelectMany(h => h.GetRuntime().Agents.AllLocallyOwnedDatabaseIds()) + .Distinct().Count().ShouldBe(2); } } diff --git a/src/Testing/CoreTests/Runtime/Agents/event_subscription_family_database_id_of.cs b/src/Testing/CoreTests/Runtime/Agents/event_subscription_family_database_id_of.cs new file mode 100644 index 000000000..2f194274f --- /dev/null +++ b/src/Testing/CoreTests/Runtime/Agents/event_subscription_family_database_id_of.cs @@ -0,0 +1,56 @@ +using System; +using JasperFx.Descriptors; +using JasperFx.Events; +using JasperFx.Events.Projections; +using Shouldly; +using Wolverine.Runtime.Agents; +using Xunit; + +namespace CoreTests.Runtime.Agents; + +// GH-3340: application code (readiness gates, per-node health checks) needs a first-class, public way to +// map an event-subscription agent URI back to the shard database it belongs to, without re-implementing +// the internal URI grammar. EventSubscriptionAgentFamily.DatabaseIdOf is that seam. +public class event_subscription_family_database_id_of +{ + [Fact] + public void extracts_the_database_id_from_a_per_tenant_agent_uri() + { + // event-subscriptions://{type}/{name}/{databaseId}/{shard...} — databaseId is DatabaseId.ToString() + var uri = EventSubscriptionAgentFamily.UriFor( + new EventStoreIdentity("main", "Marten"), + new DatabaseId("shardserver", "shard_17"), + ShardName.Compose("Trip", "All", "alpha1")); + + var id = EventSubscriptionAgentFamily.DatabaseIdOf(uri); + + id.ShouldBe(new DatabaseId("shardserver", "shard_17")); + } + + [Fact] + public void round_trips_a_realistic_sharded_database_id() + { + // The production shape: a dotted server host and a plain database name (e.g. one physical + // Postgres database per shard). This is what BuildAgentAsync parses back, and DatabaseIdOf + // must agree with it exactly. + var original = new DatabaseId("shard-cluster.internal:5432", "shard_017"); + var uri = EventSubscriptionAgentFamily.UriFor( + new EventStoreIdentity("main", "Marten"), + original, + ShardName.Compose("Trip", "All")); + + EventSubscriptionAgentFamily.DatabaseIdOf(uri).ShouldBe(original); + } + + [Fact] + public void returns_null_for_a_uri_that_is_not_this_scheme() + { + EventSubscriptionAgentFamily.DatabaseIdOf(new Uri("rabbitmq://queue/incoming")).ShouldBeNull(); + } + + [Fact] + public void returns_null_when_there_is_no_database_segment() + { + EventSubscriptionAgentFamily.DatabaseIdOf(new Uri("event-subscriptions://marten")).ShouldBeNull(); + } +} diff --git a/src/Wolverine/Runtime/Agents/EventSubscriptionAgentFamily.cs b/src/Wolverine/Runtime/Agents/EventSubscriptionAgentFamily.cs index 40942bc10..f5446d478 100644 --- a/src/Wolverine/Runtime/Agents/EventSubscriptionAgentFamily.cs +++ b/src/Wolverine/Runtime/Agents/EventSubscriptionAgentFamily.cs @@ -240,6 +240,24 @@ private async ValueTask> RetireSupersededAgentsAsync(AssignmentGrid return retired; } + /// + /// Extract the of the shard database that an event-subscription agent + /// belongs to. The URI grammar (see ) lives here, so + /// application code that needs to know which database an agent tracks should call this rather than + /// decomposing the URI segments by hand. Returns null for any URI that is not an + /// event-subscriptions:// agent URI or that does not carry a parseable database segment. + /// See GH-3340. + /// + public static DatabaseId? DatabaseIdOf(Uri uri) + { + if (uri.Scheme != SchemeName || uri.Segments.Length < 3) + { + return null; + } + + return DatabaseId.TryParse(uri.Segments[2].Trim('/'), out var id) ? id : null; + } + // 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. diff --git a/src/Wolverine/Runtime/IAgentRuntime.cs b/src/Wolverine/Runtime/IAgentRuntime.cs index e15c606fc..936783564 100644 --- a/src/Wolverine/Runtime/IAgentRuntime.cs +++ b/src/Wolverine/Runtime/IAgentRuntime.cs @@ -1,3 +1,4 @@ +using JasperFx.Descriptors; using Wolverine.Runtime.Agents; namespace Wolverine.Runtime; @@ -12,6 +13,17 @@ public interface IAgentRuntime Task InvokeAsync(NodeDestination destination, IAgentCommand command) where T : class; Uri[] AllRunningAgentUris(); + /// + /// The distinct set of shard databases whose event-subscription / projection agents are currently + /// assigned to and running on THIS node under Wolverine-managed event subscription distribution + /// (UseWolverineManagedEventSubscriptionDistribution = true). Under a + /// MultiTenantedWithShardedDatabases store with database-affine assignment, a node only runs + /// the daemon for the databases it owns; use this to scope per-node work (readiness gates, health + /// checks, progress queries) to exactly those databases instead of fanning out a connection pool to + /// every shard. Returns an empty list on nodes that own no such agents. See GH-3340. + /// + IReadOnlyList AllLocallyOwnedDatabaseIds(); + /// /// Use with caution! This will force Wolverine into restarting its leadership /// election and agent assignment diff --git a/src/Wolverine/Runtime/WolverineRuntime.Agents.cs b/src/Wolverine/Runtime/WolverineRuntime.Agents.cs index db401390f..022eb2e73 100644 --- a/src/Wolverine/Runtime/WolverineRuntime.Agents.cs +++ b/src/Wolverine/Runtime/WolverineRuntime.Agents.cs @@ -1,5 +1,6 @@ using ImTools; using JasperFx.Core; +using JasperFx.Descriptors; using Microsoft.Extensions.Logging; using Wolverine.Persistence.Durability; using Wolverine.Runtime.Agents; @@ -90,6 +91,16 @@ public Uri[] AllRunningAgentUris() return NodeController?.AllRunningAgentUris() ?? Array.Empty(); } + public IReadOnlyList AllLocallyOwnedDatabaseIds() + { + return AllRunningAgentUris() + .Select(EventSubscriptionAgentFamily.DatabaseIdOf) + .Where(id => id != null) + .Select(id => id!) + .Distinct() + .ToList(); + } + public bool IsLeader() { if (Options.Durability.Mode == DurabilityMode.Balanced && NodeController != null)