diff --git a/docs/guide/testing.md b/docs/guide/testing.md index 44defc6b6..31471b73c 100644 --- a/docs/guide/testing.md +++ b/docs/guide/testing.md @@ -207,6 +207,38 @@ await host.TrackActivity().Timeout(30.Seconds()) snippet source | anchor +### Forcing projection catch-up outside a tracked session + +Under `UseWolverineManagedEventSubscriptionDistribution` the Marten store runs in +`DaemonMode.ExternallyManaged`, so Marten's own `IHost.ForceAllMartenDaemonActivityToCatchUpAsync()` is +deliberately a **passive** read-only wait — it will not drive a paused daemon forward, and it leaves active +catch-up to the external coordinator, which is Wolverine. Use Wolverine's own helper instead: + +```csharp +// Drive every projection and subscription up to the current high water mark, +// then leave the daemons running +await host.PauseThenCatchUpOnMartenDaemonActivityAsync(); + +// ...or leave them paused, so nothing races your assertions +await host.PauseThenCatchUpOnMartenDaemonActivityAsync(CatchUpMode.AndDoNothing); + +// Ancillary stores take the store type +await host.PauseThenCatchUpOnMartenDaemonActivityAsync(); +``` + +This is the standalone equivalent of the `PauseThenCatchUpOnMartenDaemonActivity()` tracked-session stage +shown above, for suites that are not driving the work through `TrackActivity()`. It covers every +locally reachable daemon, including per-(database, tenant) shards. + +::: warning +Do **not** hand-roll this by pausing the coordinator and then calling `IProjectionDaemon.CatchUpAsync()` on +each daemon. That introduces a *second* writer of each projection's progression row alongside the shard's +own agent, which is what produces intermittent `ProgressionProgressOutOfOrderException` ("multiple processes +try to process the projection") and `23505: duplicate key value violates unique constraint +"pk_mt_event_progression"` errors. Wolverine's helper never calls `CatchUpAsync` — it resumes the agents +that already own those shards and waits for non-stale data, so there is only ever one writer. +::: + The samples shown above inlcude `Sent` message records, but there are more properties available in the `TrackedSession` object. In accordance with the `MessageEventType` enum, you can access these properties on the `TrackedSession` object: diff --git a/src/Persistence/MartenTests/TestHelpers/standalone_catch_up_under_wolverine_distribution.cs b/src/Persistence/MartenTests/TestHelpers/standalone_catch_up_under_wolverine_distribution.cs new file mode 100644 index 000000000..b2ce7b804 --- /dev/null +++ b/src/Persistence/MartenTests/TestHelpers/standalone_catch_up_under_wolverine_distribution.cs @@ -0,0 +1,140 @@ +using IntegrationTests; +using JasperFx.Core; +using JasperFx.Events.Daemon; +using JasperFx.Events.Projections; +using Marten; +using Marten.Events; +using Microsoft.Extensions.Hosting; +using Shouldly; +using Wolverine; +using Wolverine.Marten; + +namespace MartenTests.TestHelpers; + +/// +/// GH-3697: under UseWolverineManagedEventSubscriptionDistribution the Marten store runs in +/// DaemonMode.ExternallyManaged, so Marten's own ForceAllMartenDaemonActivityToCatchUpAsync +/// degrades to a passive read-only wait (marten#4904) and explicitly punts active catch-up to the external +/// coordinator. Wolverine already implemented that path, but only as a TrackActivity() stage +/// (PauseThenCatchUpOnMartenDaemonActivity). A suite that is not driving the work through a tracked +/// session had nothing to call, so every one of them reimplemented "pause the coordinator, then +/// StopAllAsync + CatchUpAsync each daemon" by hand — which races the coordinator and throws +/// ProgressionProgressOutOfOrderException and duplicate-key errors on pk_mt_event_progression. +/// +/// These cover the standalone entry point. It never calls CatchUpAsync: the agents that already own +/// the shards are resumed and drained, so there is only ever one writer of each progression row. +/// +public class standalone_catch_up_under_wolverine_distribution : IAsyncLifetime +{ + private IHost _host = null!; + + public async ValueTask InitializeAsync() + { + _host = await Host.CreateDefaultBuilder() + .UseWolverine(opts => + { + opts.Services.AddMarten(m => + { + m.DisableNpgsqlLogging = true; + m.Connection(Servers.PostgresConnectionString); + m.DatabaseSchemaName = "letters7"; + + m.Projections.Add(ProjectionLifecycle.Async); + }).IntegrateWithWolverine(x => x.UseWolverineManagedEventSubscriptionDistribution = true); + + opts.Services.AddMartenStore(m => + { + m.DisableNpgsqlLogging = true; + m.Connection(Servers.PostgresConnectionString); + m.DatabaseSchemaName = "letters8"; + + m.Projections.Add(ProjectionLifecycle.Async); + }).IntegrateWithWolverine(); + + opts.Durability.Mode = DurabilityMode.Solo; + }).StartAsync(); + } + + public async ValueTask DisposeAsync() + { + await _host.StopAsync(); + _host.Dispose(); + } + + [Fact] + public async Task catches_up_the_main_store_with_no_tracked_session() + { + await _host.ResetAllMartenDataAsync(); + + var id = Guid.NewGuid(); + + await using var session = _host.DocumentStore().LightweightSession(); + session.Events.StartStream(id, "AAAACCCCBDEEE".ToLetterEvents()); + await session.SaveChangesAsync(TestContext.Current.CancellationToken); + + await _host.PauseThenCatchUpOnMartenDaemonActivityAsync( + cancellation: TestContext.Current.CancellationToken); + + var counts = await session.LoadAsync(id, TestContext.Current.CancellationToken); + counts.ShouldNotBeNull(); + counts.ACount.ShouldBe(4); + counts.BCount.ShouldBe(1); + counts.CCount.ShouldBe(4); + } + + [Fact] + public async Task catches_up_an_ancillary_store_with_no_tracked_session() + { + await _host.ResetAllMartenDataAsync(); + + var id = Guid.NewGuid(); + + await using var session = _host.DocumentStore().LightweightSession(); + session.Events.StartStream(id, "AAAACCCCBDEEE".ToLetterEvents()); + await session.SaveChangesAsync(TestContext.Current.CancellationToken); + + await _host.PauseThenCatchUpOnMartenDaemonActivityAsync( + cancellation: TestContext.Current.CancellationToken); + + var counts = await session.LoadAsync(id, TestContext.Current.CancellationToken); + counts.ShouldNotBeNull(); + counts.ACount.ShouldBe(4); + counts.CCount.ShouldBe(4); + } + + /// + /// AndDoNothing has to leave the coordinator paused, so a suite can catch up, assert, append more events, + /// and catch up again without the daemon racing its assertions in between. + /// + [Fact] + public async Task and_do_nothing_leaves_the_daemons_paused_and_a_second_pass_still_works() + { + await _host.ResetAllMartenDataAsync(); + + var id = Guid.NewGuid(); + + await using var session = _host.DocumentStore().LightweightSession(); + session.Events.StartStream(id, "AAAA".ToLetterEvents()); + await session.SaveChangesAsync(TestContext.Current.CancellationToken); + + await _host.PauseThenCatchUpOnMartenDaemonActivityAsync(CatchUpMode.AndDoNothing, + cancellation: TestContext.Current.CancellationToken); + + (await session.LoadAsync(id, TestContext.Current.CancellationToken))! + .ACount.ShouldBe(4); + + // Appended while the coordinator is paused -- nothing should be projecting it yet. + session.Events.Append(id, "BBB".ToLetterEvents()); + await session.SaveChangesAsync(TestContext.Current.CancellationToken); + await Task.Delay(500.Milliseconds(), TestContext.Current.CancellationToken); + + // A paused coordinator hands back no daemons to drive, which is exactly the state the hand-rolled + // workaround in the report was fighting. The supported path resumes them instead. + await _host.PauseThenCatchUpOnMartenDaemonActivityAsync(CatchUpMode.AndDoNothing, + cancellation: TestContext.Current.CancellationToken); + + var counts = await session.LoadAsync(id, TestContext.Current.CancellationToken); + counts!.ACount.ShouldBe(4); + counts.BCount.ShouldBe(3); + } +} diff --git a/src/Persistence/Wolverine.Marten/TestingExtensions.cs b/src/Persistence/Wolverine.Marten/TestingExtensions.cs index b6f953d4d..64c0a8e95 100644 --- a/src/Persistence/Wolverine.Marten/TestingExtensions.cs +++ b/src/Persistence/Wolverine.Marten/TestingExtensions.cs @@ -5,6 +5,7 @@ using Marten.Events; using Marten.Events.Daemon.Coordination; using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Hosting; using Wolverine.Runtime; using Wolverine.Tracking; @@ -122,6 +123,115 @@ await catchUpThroughCoordinatorAsync(runtime, envelope, coordinator, private static readonly TimeSpan CatchUpTimeout = TimeSpan.FromSeconds(60); + /// + /// Force every Marten projection and subscription running asynchronously under Wolverine-managed event + /// subscription distribution to catch up to the current high water mark, then restore the daemons to the + /// state named by . + /// + /// This is the standalone equivalent of + /// , for suites + /// that are not driving the work through TrackActivity(). Marten's own + /// IHost.ForceAllMartenDaemonActivityToCatchUpAsync() is deliberately a passive read-only wait when + /// the store is DaemonMode.ExternallyManaged — which is the mode Wolverine puts it in — and punts + /// active catch-up to the external coordinator. This is that path. + /// + /// Note what it does NOT do: it never calls IProjectionDaemon.CatchUpAsync. Doing that while the + /// coordinator owns the shards introduces a *second* writer of each progression row, which is what produces + /// the ProgressionProgressOutOfOrderException ("multiple processes try to process the projection") + /// and the 23505 duplicate key value violates unique constraint "pk_mt_event_progression" on a + /// shard's first progression row. Resuming the agents that already own those shards and waiting for + /// non-stale data means there is only ever one writer, so neither race exists to be swallowed. + /// + /// + /// Whether to leave the daemons running afterwards (the default) or re-pause them + /// How long to allow for catch-up. Defaults to 60 seconds. + /// + public static Task PauseThenCatchUpOnMartenDaemonActivityAsync(this IHost host, + CatchUpMode mode = CatchUpMode.AndResumeNormally, TimeSpan? timeout = null, + CancellationToken cancellation = default) + { + return host.Services.PauseThenCatchUpOnMartenDaemonActivityAsync(mode, timeout, cancellation); + } + + /// + /// Force every Marten projection and subscription running asynchronously under Wolverine-managed event + /// subscription distribution to catch up to the current high water mark, then restore the daemons to the + /// state named by . See the IHost overload for the full rationale. + /// + public static Task PauseThenCatchUpOnMartenDaemonActivityAsync(this IServiceProvider services, + CatchUpMode mode = CatchUpMode.AndResumeNormally, TimeSpan? timeout = null, + CancellationToken cancellation = default) + { + var coordinator = services.GetRequiredService(); + + return resumeAndWaitForNonStaleAsync(coordinator, + t => services.DocumentStore().WaitForNonStaleProjectionDataAsync(t), + mode, timeout ?? CatchUpTimeout, "the main Marten store", cancellation); + } + + /// + /// Force every Marten projection and subscription in an ancillary store running asynchronously under + /// Wolverine-managed event subscription distribution to catch up to the current high water mark, then + /// restore the daemons to the state named by . See the non-generic IHost + /// overload for the full rationale. + /// + public static Task PauseThenCatchUpOnMartenDaemonActivityAsync(this IHost host, + CatchUpMode mode = CatchUpMode.AndResumeNormally, TimeSpan? timeout = null, + CancellationToken cancellation = default) where T : class, IDocumentStore + { + return host.Services.PauseThenCatchUpOnMartenDaemonActivityAsync(mode, timeout, cancellation); + } + + /// + /// Force every Marten projection and subscription in an ancillary store running asynchronously under + /// Wolverine-managed event subscription distribution to catch up to the current high water mark, then + /// restore the daemons to the state named by . See the non-generic IHost + /// overload for the full rationale. + /// + public static Task PauseThenCatchUpOnMartenDaemonActivityAsync(this IServiceProvider services, + CatchUpMode mode = CatchUpMode.AndResumeNormally, TimeSpan? timeout = null, + CancellationToken cancellation = default) where T : class, IDocumentStore + { + var coordinator = services.GetRequiredService>(); + + return resumeAndWaitForNonStaleAsync(coordinator, + t => services.DocumentStore().WaitForNonStaleProjectionDataAsync(t), + mode, timeout ?? CatchUpTimeout, $"the ancillary Marten store {typeof(T).NameInCode()}", cancellation); + } + + // The whole of "force catch-up" under a Wolverine-managed coordinator: make sure the agents that own the + // shards are running, wait for every (database, tenant) shard to reach the current high water mark, then + // honor the caller's CatchUpMode. Same algorithm the tracked-session stage runs in + // catchUpThroughCoordinatorAsync below, minus the message-tracking bookkeeping that only makes sense + // inside a TrackedSession. + private static async Task resumeAndWaitForNonStaleAsync( + global::Marten.Events.Daemon.Coordination.IProjectionCoordinator coordinator, + Func waitForNonStale, + CatchUpMode mode, + TimeSpan timeout, + string storeDescription, + CancellationToken cancellation) + { + await coordinator.ResumeAsync().ConfigureAwait(false); + + try + { + await waitForNonStale(timeout).WaitAsync(cancellation).ConfigureAwait(false); + } + catch (OperationCanceledException) when (cancellation.IsCancellationRequested) + { + throw new TimeoutException( + $"Cancelled before the Marten projections for {storeDescription} caught up. The catch-up itself was allowed {timeout.TotalSeconds:N0} seconds."); + } + finally + { + if (mode == CatchUpMode.AndDoNothing) + { + await coordinator.PauseAsync().ConfigureAwait(false); + } + } + } + // GH-3349 / Marten #4904: under Wolverine-managed event-subscription distribution the Marten store is // DaemonMode.ExternallyManaged, so Marten's ForceAllMartenDaemonActivityToCatchUpAsync no longer DRIVES // a paused daemon forward — it degrades to a passive wait for non-stale data (to avoid a