Standalone force-catch-up for Marten under Wolverine-managed distribution (GH-3697)

docs/guide/testing.md

@@ -207,6 +207,38 @@ await host.TrackActivity().Timeout(30.Seconds())
 <sup><a href='https://github.com/JasperFx/wolverine/blob/main/src/Samples/DocumentationSamples/TestingSupportSamples.cs#L136-L198' title='Snippet source file'>snippet source</a> | <a href='#snippet-sample_advanced_tracked_session_usage' title='Start of snippet'>anchor</a></sup>
 <!-- endSnippet -->
 
+### 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<ILetterStore>();
+```
+
+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:
 

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;
+
+/// <summary>
+/// GH-3697: under <c>UseWolverineManagedEventSubscriptionDistribution</c> the Marten store runs in
+/// <c>DaemonMode.ExternallyManaged</c>, so Marten's own <c>ForceAllMartenDaemonActivityToCatchUpAsync</c>
+/// 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 <c>TrackActivity()</c> stage
+/// (<c>PauseThenCatchUpOnMartenDaemonActivity</c>). 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
+/// <c>StopAllAsync</c> + <c>CatchUpAsync</c> each daemon" by hand — which races the coordinator and throws
+/// <c>ProgressionProgressOutOfOrderException</c> and duplicate-key errors on <c>pk_mt_event_progression</c>.
+///
+/// These cover the standalone entry point. It never calls <c>CatchUpAsync</c>: the agents that already own
+/// the shards are resumed and drained, so there is only ever one writer of each progression row.
+/// </summary>
+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<LetterCountsProjection>(ProjectionLifecycle.Async);
+                }).IntegrateWithWolverine(x => x.UseWolverineManagedEventSubscriptionDistribution = true);
+
+                opts.Services.AddMartenStore<ILetterStore>(m =>
+                {
+                    m.DisableNpgsqlLogging = true;
+                    m.Connection(Servers.PostgresConnectionString);
+                    m.DatabaseSchemaName = "letters8";
+
+                    m.Projections.Add<LetterCountsProjection>(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<LetterCounts>(id, "AAAACCCCBDEEE".ToLetterEvents());
+        await session.SaveChangesAsync(TestContext.Current.CancellationToken);
+
+        await _host.PauseThenCatchUpOnMartenDaemonActivityAsync(
+            cancellation: TestContext.Current.CancellationToken);
+
+        var counts = await session.LoadAsync<LetterCounts>(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<ILetterStore>();
+
+        var id = Guid.NewGuid();
+
+        await using var session = _host.DocumentStore<ILetterStore>().LightweightSession();
+        session.Events.StartStream<LetterCounts>(id, "AAAACCCCBDEEE".ToLetterEvents());
+        await session.SaveChangesAsync(TestContext.Current.CancellationToken);
+
+        await _host.PauseThenCatchUpOnMartenDaemonActivityAsync<ILetterStore>(
+            cancellation: TestContext.Current.CancellationToken);
+
+        var counts = await session.LoadAsync<LetterCounts>(id, TestContext.Current.CancellationToken);
+        counts.ShouldNotBeNull();
+        counts.ACount.ShouldBe(4);
+        counts.CCount.ShouldBe(4);
+    }
+
+    /// <summary>
+    /// 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.
+    /// </summary>
+    [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<LetterCounts>(id, "AAAA".ToLetterEvents());
+        await session.SaveChangesAsync(TestContext.Current.CancellationToken);
+
+        await _host.PauseThenCatchUpOnMartenDaemonActivityAsync(CatchUpMode.AndDoNothing,
+            cancellation: TestContext.Current.CancellationToken);
+
+        (await session.LoadAsync<LetterCounts>(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<LetterCounts>(id, TestContext.Current.CancellationToken);
+        counts!.ACount.ShouldBe(4);
+        counts.BCount.ShouldBe(3);
+    }
+}

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);
 
+    /// <summary>
+    /// 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 <paramref name="mode"/>.
+    ///
+    /// This is the standalone equivalent of
+    /// <see cref="PauseThenCatchUpOnMartenDaemonActivity(TrackedSessionConfiguration,CatchUpMode)"/>, for suites
+    /// that are not driving the work through <c>TrackActivity()</c>. Marten's own
+    /// <c>IHost.ForceAllMartenDaemonActivityToCatchUpAsync()</c> is deliberately a passive read-only wait when
+    /// the store is <c>DaemonMode.ExternallyManaged</c> — 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 <c>IProjectionDaemon.CatchUpAsync</c>. Doing that while the
+    /// coordinator owns the shards introduces a *second* writer of each progression row, which is what produces
+    /// the <c>ProgressionProgressOutOfOrderException</c> ("multiple processes try to process the projection")
+    /// and the <c>23505 duplicate key value violates unique constraint "pk_mt_event_progression"</c> 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.
+    /// </summary>
+    /// <param name="host"></param>
+    /// <param name="mode">Whether to leave the daemons running afterwards (the default) or re-pause them</param>
+    /// <param name="timeout">How long to allow for catch-up. Defaults to 60 seconds.</param>
+    /// <param name="cancellation"></param>
+    public static Task PauseThenCatchUpOnMartenDaemonActivityAsync(this IHost host,
+        CatchUpMode mode = CatchUpMode.AndResumeNormally, TimeSpan? timeout = null,
+        CancellationToken cancellation = default)
+    {
+        return host.Services.PauseThenCatchUpOnMartenDaemonActivityAsync(mode, timeout, cancellation);
+    }
+
+    /// <summary>
+    /// 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 <paramref name="mode"/>. See the <c>IHost</c> overload for the full rationale.
+    /// </summary>
+    public static Task PauseThenCatchUpOnMartenDaemonActivityAsync(this IServiceProvider services,
+        CatchUpMode mode = CatchUpMode.AndResumeNormally, TimeSpan? timeout = null,
+        CancellationToken cancellation = default)
+    {
+        var coordinator = services.GetRequiredService<IProjectionCoordinator>();
+
+        return resumeAndWaitForNonStaleAsync(coordinator,
+            t => services.DocumentStore().WaitForNonStaleProjectionDataAsync(t),
+            mode, timeout ?? CatchUpTimeout, "the main Marten store", cancellation);
+    }
+
+    /// <summary>
+    /// 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 <paramref name="mode"/>. See the non-generic <c>IHost</c>
+    /// overload for the full rationale.
+    /// </summary>
+    public static Task PauseThenCatchUpOnMartenDaemonActivityAsync<T>(this IHost host,
+        CatchUpMode mode = CatchUpMode.AndResumeNormally, TimeSpan? timeout = null,
+        CancellationToken cancellation = default) where T : class, IDocumentStore
+    {
+        return host.Services.PauseThenCatchUpOnMartenDaemonActivityAsync<T>(mode, timeout, cancellation);
+    }
+
+    /// <summary>
+    /// 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 <paramref name="mode"/>. See the non-generic <c>IHost</c>
+    /// overload for the full rationale.
+    /// </summary>
+    public static Task PauseThenCatchUpOnMartenDaemonActivityAsync<T>(this IServiceProvider services,
+        CatchUpMode mode = CatchUpMode.AndResumeNormally, TimeSpan? timeout = null,
+        CancellationToken cancellation = default) where T : class, IDocumentStore
+    {
+        var coordinator = services.GetRequiredService<IProjectionCoordinator<T>>();
+
+        return resumeAndWaitForNonStaleAsync(coordinator,
+            t => services.DocumentStore<T>().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<TimeSpan, Task> 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