From 2a311d7731378d33100ba38f3feab35da63042b7 Mon Sep 17 00:00:00 2001 From: Anne Erdtsieck Date: Tue, 30 Jun 2026 10:49:48 +0200 Subject: [PATCH 1/3] Surface DistributesAgentsPerTenant for sharded + per-tenant-partitioned stores Overrides IEventStore.DistributesAgentsPerTenant => UseTenantPartitionedEvents && Tenancy is ShardedTenancy, so Wolverine-managed distribution fans out per-tenant agents only where tenants are co-located in a shard DB with independent sequences. Single-DB partitioning and database-per-tenant stay one agent per database. See JasperFx/wolverine#3280. --- src/Marten/DocumentStore.EventStore.cs | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/src/Marten/DocumentStore.EventStore.cs b/src/Marten/DocumentStore.EventStore.cs index 19d3c64526..c4edb6afb3 100644 --- a/src/Marten/DocumentStore.EventStore.cs +++ b/src/Marten/DocumentStore.EventStore.cs @@ -65,6 +65,14 @@ bool IEventStore.HasMultipleTenants } } + // wolverine#3280: under sharded databases + per-tenant event partitioning, multiple tenants are + // co-located in one shard database and each draws its own mt_events_sequence_ (independent, + // overlapping). A single store-global high-water cannot track them, so node-distributed daemons must + // fan out one agent per (shard, tenant). Single-database partitioning and database-per-tenant do NOT + // need this (one agent per database already covers a single tenant or the single partitioned table). + bool IEventStore.DistributesAgentsPerTenant + => Options.Events.UseTenantPartitionedEvents && Options.Tenancy is Storage.ShardedTenancy; + async ValueTask> IEventStore.AllDatabases() { // Straight delegation to ITenancy, mirroring IMartenStorage.AllDatabases(). The IMartenDatabase From 0cddc62364bdce1689917eed19a1234e208030ab Mon Sep 17 00:00:00 2001 From: "Jeremy D. Miller" Date: Mon, 6 Jul 2026 11:28:22 -0500 Subject: [PATCH 2/3] Tests: per-tenant agent start routing + DistributesAgentsPerTenant gating Restores the regression test from #4799. per_tenant_agent_start_routes_high_water reproduces the Wolverine-managed shape (StartAgentAsync per tenant shard, no following store-global high-water tick) and requires the hardened per-tenant StartAgentAsync from jasperfx#487. distributes_agents_per_tenant_gating extracts the DistributesAgentsPerTenant assertions from the archived affine-gating tests, adjusted to compile without the dropped database-affine surface (GroupAgentAssignmentsByDatabase / UseDatabaseAffineAgentAssignment / MaxNodesPerDatabaseForAgents). Co-Authored-By: Anne Erdtsieck Co-Authored-By: Claude Fable 5 --- .../distributes_agents_per_tenant_gating.cs | 70 ++++++++++ ...er_tenant_agent_start_routes_high_water.cs | 130 ++++++++++++++++++ 2 files changed, 200 insertions(+) create mode 100644 src/TenantPartitionedEventsTests/Config/distributes_agents_per_tenant_gating.cs create mode 100644 src/TenantPartitionedEventsTests/Regressions/per_tenant_agent_start_routes_high_water.cs diff --git a/src/TenantPartitionedEventsTests/Config/distributes_agents_per_tenant_gating.cs b/src/TenantPartitionedEventsTests/Config/distributes_agents_per_tenant_gating.cs new file mode 100644 index 0000000000..30316a68a1 --- /dev/null +++ b/src/TenantPartitionedEventsTests/Config/distributes_agents_per_tenant_gating.cs @@ -0,0 +1,70 @@ +using JasperFx.Events; +using Marten; +using Marten.Events; +using Marten.Testing.Harness; +using Shouldly; +using Xunit; + +namespace TenantPartitionedEventsTests.Config; + +/// +/// wolverine#3280 — how is gated. It only turns on +/// for a sharded per-tenant-partitioned store (the one tenancy where multiple tenants are co-located in a +/// shard database with independent per-tenant sequences, so node-distributed daemons must fan agents out +/// per (shard, tenant)). Everywhere else it stays false so distribution behaves exactly as before: one +/// agent per database. These are pure configuration assertions — building the store never opens a +/// connection. +/// +public class distributes_agents_per_tenant_gating +{ + [Fact] + public void off_on_a_single_database_store_even_with_per_tenant_partitioning() + { + using var store = DocumentStore.For(opts => + { + opts.Connection(ConnectionSource.ConnectionString); + opts.Events.TenancyStyle = TenancyStyle.Conjoined; + opts.Events.UseTenantPartitionedEvents = true; + }); + + ((IEventStore)store).DistributesAgentsPerTenant + .ShouldBeFalse("a single-database store keeps one agent per database"); + } + + [Fact] + public void on_for_a_sharded_per_tenant_partitioned_store() + { + using var store = ShardedStore(partitioned: true); + + ((IEventStore)store).DistributesAgentsPerTenant + .ShouldBeTrue("sharded + per-tenant partitioning fans agents out per tenant"); + } + + [Fact] + public void off_for_a_sharded_store_without_per_tenant_partitioning() + { + using var store = ShardedStore(partitioned: false); + + ((IEventStore)store).DistributesAgentsPerTenant + .ShouldBeFalse("without per-tenant partitioning a shard database has a single sequence"); + } + + // A sharded store with two databases. Dummy connection strings — the assertions only read + // StoreOptions, so the store is never connected or provisioned. + private static IDocumentStore ShardedStore(bool partitioned) => + DocumentStore.For(opts => + { + opts.MultiTenantedWithShardedDatabases(x => + { + x.ConnectionString = ConnectionSource.ConnectionString; + x.SchemaName = "distributes_gating"; + x.PartitionSchemaName = "distributes_gating_tenants"; + x.AddDatabase("db1", ConnectionSource.ConnectionString); + x.AddDatabase("db2", ConnectionSource.ConnectionString); + }); + + opts.Events.TenancyStyle = TenancyStyle.Conjoined; + opts.Events.AppendMode = EventAppendMode.QuickWithServerTimestamps; + opts.Events.UseTenantPartitionedEvents = partitioned; + }); +} diff --git a/src/TenantPartitionedEventsTests/Regressions/per_tenant_agent_start_routes_high_water.cs b/src/TenantPartitionedEventsTests/Regressions/per_tenant_agent_start_routes_high_water.cs new file mode 100644 index 0000000000..c641237a01 --- /dev/null +++ b/src/TenantPartitionedEventsTests/Regressions/per_tenant_agent_start_routes_high_water.cs @@ -0,0 +1,130 @@ +#nullable enable +using System; +using System.Collections.Generic; +using System.Linq; +using System.Threading; +using System.Threading.Tasks; +using JasperFx; +using JasperFx.Events; +using JasperFx.Events.Aggregation; +using JasperFx.Events.Projections; +using Marten; +using Marten.Events; +using Marten.Events.Aggregation; +using Marten.Events.Projections; +using Marten.Storage; +using Marten.Testing.Harness; +using Shouldly; +using Xunit; + +namespace TenantPartitionedEventsTests.Regressions; + +/// +/// wolverine#3280 / JasperFx per-tenant agent start: under Wolverine-managed event-subscription +/// distribution the daemon is driven by StartAgentAsync("<proj>:V{n}:All:<tenant>") for +/// each per-tenant shard individually — NOT by StartAllAsync (which primes every tenant's ceiling +/// up front). A per-tenant agent therefore starts with its ceiling unknown (high-water 0), and the +/// per-tenant high-water poll only re-runs on a store-global high-water CHANGE tick. When a node starts +/// after the store-global mark is already stable — the common case for a node that joins after catch-up — +/// no further tick fires and the freshly-started agent would sit idle at 0 forever. +/// +/// +/// The fix (JasperFxAsyncDaemon.StartAgentAsync) drives one per-tenant poll immediately after a +/// per-tenant agent starts, so it is routed its own tenant's mark and advances. This test reproduces the +/// "start individually, no following global tick" shape: append events for two tenants, then start each +/// tenant's async agent one-by-one via the string overload (exactly what Wolverine calls) and assert both +/// tenants' projections catch up. It composes with the per-tenant high-water reading from max(seq_id) +/// (#4712-class fix) so the mark is correct even though the per-tenant sequence is never advanced. +/// +/// +public partial class per_tenant_agent_start_routes_high_water +{ + private static readonly string SchemaName = $"pt_agent_start_p{Environment.ProcessId}"; + + public class TenantTripDistance + { + public Guid Id { get; set; } + public double Distance { get; set; } + public int Version { get; set; } + } + + public record TripStarted(Guid Id); + public record TripLeg(double Distance); + + public partial class TenantTripDistanceProjection: SingleStreamProjection + { + public TenantTripDistanceProjection() => Name = "PtAgentStartTrip"; + public void Apply(TenantTripDistance agg, TripLeg @event) => agg.Distance += @event.Distance; + } + + [Fact] + public async Task starting_each_per_tenant_agent_individually_advances_its_projection() + { + using var store = DocumentStore.For(opts => + { + opts.Connection(ConnectionSource.ConnectionString); + opts.DatabaseSchemaName = SchemaName; + opts.Events.TenancyStyle = TenancyStyle.Conjoined; + opts.Events.UseTenantPartitionedEvents = true; + opts.Events.AppendMode = EventAppendMode.QuickWithServerTimestamps; + opts.Policies.AllDocumentsAreMultiTenanted(); + opts.Projections.Add(ProjectionLifecycle.Async); + // Nested type name + tenant suffix overflows PG's 64-byte identifier limit; shorten it. + opts.Schema.For().Identity(x => x.Id).DocumentAlias("pt_agent_trip"); + }); + await store.Advanced.Clean.CompletelyRemoveAllAsync(); + await store.Storage.Database.EnsureStorageExistsAsync(typeof(IEvent)); + + var tenants = new[] { $"t_{Guid.NewGuid():N}"[..12], $"t_{Guid.NewGuid():N}"[..12] }; + await store.Advanced.AddMartenManagedTenantsAsync(CancellationToken.None, tenants); + + // A distinct stream per tenant, each with one TripLeg (Distance 1.0). + var streamByTenant = new Dictionary(); + foreach (var tenant in tenants) + { + var streamId = Guid.NewGuid(); + streamByTenant[tenant] = streamId; + await using var session = store.LightweightSession(tenant); + session.Events.StartStream(streamId, new TripStarted(streamId), new TripLeg(1.0)); + await session.SaveChangesAsync(); + } + + using var daemon = await store.BuildProjectionDaemonAsync(); + using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(30)); + + // Start the FIRST tenant's agent and let it fully catch up. Starting it also starts high-water + // detection, which advances the store-global mark to its final value and broadcasts it. + var first = tenants[0]; + await daemon.StartAgentAsync(ShardName.Compose("PtAgentStartTrip", ShardName.All, first).Identity, CancellationToken.None); + await WaitForProjectionAsync(store, first, streamByTenant[first], cts.Token); + + // Now start the SECOND tenant's agent. The store-global high-water is already at its final value, so + // HighWaterAgent will NOT broadcast again (it publishes only on change — HighWaterAgent.cs:209), and + // no store-global tick will ever fire the per-tenant poll for this agent. It therefore advances ONLY + // if StartAgentAsync itself drives a per-tenant poll after starting it (the fix). Under the pre-fix + // behavior it sits at high-water 0 forever and this wait times out — the deterministic regression. + var second = tenants[1]; + await daemon.StartAgentAsync(ShardName.Compose("PtAgentStartTrip", ShardName.All, second).Identity, CancellationToken.None); + await WaitForProjectionAsync(store, second, streamByTenant[second], cts.Token); + } + + private static async Task WaitForProjectionAsync( + IDocumentStore store, string tenant, Guid streamId, CancellationToken token) + { + while (!token.IsCancellationRequested) + { + await using var query = store.QuerySession(tenant); + var doc = await query.LoadAsync(streamId, token); + if (doc is { Distance: >= 1.0 }) + { + return; + } + + await Task.Delay(150, token); + } + + throw new Xunit.Sdk.XunitException( + $"per-tenant agent for {tenant} never advanced its projection (stream {streamId}); " + + "its own high-water mark was not routed after StartAgentAsync."); + } +} From 0ab895bd7e4617c641388afdebf89e2ebe58e5cd Mon Sep 17 00:00:00 2001 From: "Jeremy D. Miller" Date: Mon, 6 Jul 2026 13:46:15 -0500 Subject: [PATCH 3/3] =?UTF-8?q?Consume=20JasperFx=202.20.0=20(jasperfx#487?= =?UTF-8?q?=20=E2=80=94=20per-tenant=20agent=20distribution=20surface)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Fable 5 --- Directory.Packages.props | 12 ++++++++---- 1 file changed, 8 insertions(+), 4 deletions(-) diff --git a/Directory.Packages.props b/Directory.Packages.props index 8dbc88997c..4e9f05a90e 100644 --- a/Directory.Packages.props +++ b/Directory.Packages.props @@ -73,13 +73,17 @@ AggregateType (TypeDescriptor), populated in the shared SubscriptionDescriptor(ISubscriptionSource, IEventStore) ctor that Marten's projections funnel through via ProjectionGraph.Describe. See marten#4772. --> - - - + + + + all runtime; build; native; contentfiles; analyzers; buildtransitive - +