GH-5094 Wolverine.Fisher: opt-in Wolverine-managed event subscription distribution

src/Persistence/FisherTests/fisher_managed_event_subscription_distribution_5094.cs

@@ -0,0 +1,171 @@
+using Fisher;
+using JasperFx;
+using JasperFx.Events.Daemon;
+using JasperFx.Events.Projections;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Hosting;
+using Shouldly;
+using Wolverine;
+using Wolverine.Fisher;
+using Wolverine.Runtime.Agents;
+using Wolverine.Tracking;
+using FisherCoordinator = Wolverine.Fisher.Distribution.IProjectionCoordinator;
+
+namespace FisherTests;
+
+/// <summary>
+///     GH-5094. With <see cref="FisherIntegration.UseWolverineManagedEventSubscriptionDistribution" />, a Fisher
+///     host runs its async projections as Wolverine event subscription agents, so operator tooling (CritterWatch
+///     #1437) pauses, restarts and rewinds them through the agent family the same way it does on Marten and Polecat.
+/// </summary>
+public class fisher_managed_event_subscription_distribution_5094 : IAsyncLifetime
+{
+    private FisherTestDatabase theDatabase = null!;
+    private IHost _host = null!;
+
+    public async ValueTask InitializeAsync()
+    {
+        theDatabase = Servers.CreateDatabase("managed_distribution_5094");
+
+        _host = await Host.CreateDefaultBuilder()
+            .UseWolverine(opts =>
+            {
+                opts.Discovery.DisableConventionalDiscovery();
+                opts.Durability.Mode = DurabilityMode.Solo;
+
+                opts.Services.AddFisher(m =>
+                    {
+                        m.Connection(theDatabase.ConnectionString);
+                        m.AutoCreateSchemaObjects = AutoCreate.All;
+                        m.Projections.Snapshot<ManagedTally>(SnapshotLifecycle.Async);
+                    })
+                    .ApplyAllDatabaseChangesOnStartup()
+                    .IntegrateWithWolverine(x => x.UseWolverineManagedEventSubscriptionDistribution = true);
+            }).StartAsync();
+    }
+
+    public async ValueTask DisposeAsync()
+    {
+        await _host.StopAsync();
+        _host.Dispose();
+        theDatabase.Dispose();
+    }
+
+    [Fact]
+    public void registers_the_event_subscription_agent_family()
+    {
+        _host.Services.GetServices<IEventSubscriptionAgentFamily>().ShouldHaveSingleItem();
+        _host.Services.GetServices<IAgentFamily>().OfType<EventSubscriptionAgentFamily>().ShouldHaveSingleItem();
+
+        // Wolverine runs the daemon, not Fisher's own hosted service
+        _host.Services.GetServices<IProjectionCoordinator>().ShouldBeEmpty();
+        _host.Services.GetRequiredService<FisherCoordinator>().GetType().Name.ShouldBe("WolverineProjectionCoordinator");
+    }
+
+    [Fact]
+    public async Task the_projection_is_an_agent_that_tooling_can_find()
+    {
+        var family = _host.Services.GetRequiredService<IEventSubscriptionAgentFamily>();
+
+        var uri = await family.FindAgentUriAsync("ManagedTally:All", null, TestContext.Current.CancellationToken);
+
+        uri.ShouldNotBeNull();
+        uri.Scheme.ShouldBe(EventSubscriptionAgentFamily.SchemeName);
+        uri.AbsolutePath.TrimEnd('/').ShouldEndWith("/managedtally/all");
+    }
+
+    [Fact]
+    public async Task the_agent_runs_on_this_node_and_projects()
+    {
+        var family = _host.Services.GetRequiredService<IEventSubscriptionAgentFamily>();
+        var uri = await family.FindAgentUriAsync("ManagedTally:All", null, TestContext.Current.CancellationToken);
+
+        var runtime = _host.GetRuntime();
+        await waitFor(() => runtime.Agents.AllRunningAgentUris().Contains(uri!), "the projection agent to start");
+
+        var id = Guid.NewGuid();
+        var store = _host.Services.GetRequiredService<IDocumentStore>();
+        await using (var session = store.LightweightSession())
+        {
+            session.Events.StartStream<ManagedTally>(id, new ManagedTallied(2), new ManagedTallied(3));
+            await session.SaveChangesAsync(TestContext.Current.CancellationToken);
+        }
+
+        ManagedTally? tally = null;
+        await waitFor(async () =>
+        {
+            await using var query = store.QuerySession();
+            tally = await query.LoadAsync<ManagedTally>(id, TestContext.Current.CancellationToken);
+            return tally is { Total: 5 };
+        }, "the async projection to catch up");
+    }
+
+    [Fact]
+    public async Task pausing_through_the_coordinator_stops_the_agent()
+    {
+        var family = _host.Services.GetRequiredService<IEventSubscriptionAgentFamily>();
+        var uri = await family.FindAgentUriAsync("ManagedTally:All", null, TestContext.Current.CancellationToken);
+        var runtime = _host.GetRuntime();
+        await waitFor(() => runtime.Agents.AllRunningAgentUris().Contains(uri!), "the projection agent to start");
+
+        var coordinator = _host.Services.GetRequiredService<FisherCoordinator>();
+        var daemon = (await coordinator.AllDaemonsAsync()).ShouldHaveSingleItem();
+
+        await coordinator.PauseAsync();
+        await waitFor(() => daemon.CurrentAgents().Count == 0, "the daemon to stop its shards");
+
+        await coordinator.ResumeAsync();
+        await waitFor(() => daemon.CurrentAgents().Count > 0, "the daemon to restart its shards");
+    }
+
+    [Fact]
+    public async Task refuses_fishers_own_solo_daemon_alongside_it()
+    {
+        using var database = Servers.CreateDatabase("managed_distribution_5094_solo");
+
+        var ex = await Should.ThrowAsync<InvalidOperationException>(async () =>
+        {
+            using var host = await Host.CreateDefaultBuilder()
+                .UseWolverine(opts =>
+                {
+                    opts.Discovery.DisableConventionalDiscovery();
+                    opts.Durability.Mode = DurabilityMode.Solo;
+
+                    opts.Services.AddFisher(m =>
+                        {
+                            m.Connection(database.ConnectionString);
+                            m.AutoCreateSchemaObjects = AutoCreate.All;
+                            m.Projections.Snapshot<ManagedTally>(SnapshotLifecycle.Async);
+                        })
+                        .AddAsyncDaemon(DaemonMode.Solo)
+                        .IntegrateWithWolverine(x => x.UseWolverineManagedEventSubscriptionDistribution = true);
+                }).StartAsync();
+        });
+
+        ex.Message.ShouldContain("UseWolverineManagedEventSubscriptionDistribution");
+    }
+
+    private static Task waitFor(Func<bool> condition, string what) => waitFor(() => Task.FromResult(condition()), what);
+
+    private static async Task waitFor(Func<Task<bool>> condition, string what)
+    {
+        var deadline = DateTimeOffset.UtcNow.AddSeconds(30);
+        while (DateTimeOffset.UtcNow < deadline)
+        {
+            if (await condition()) return;
+            await Task.Delay(100);
+        }
+
+        throw new TimeoutException($"Timed out waiting for {what}");
+    }
+}
+
+public record ManagedTallied(int Amount);
+
+public class ManagedTally
+{
+    public Guid Id { get; set; }
+    public int Total { get; set; }
+
+    public void Apply(ManagedTallied e) => Total += e.Amount;
+}

src/Persistence/Wolverine.Fisher/AncillaryWolverineOptionsFisherExtensions.cs

@@ -5,6 +5,8 @@
 using JasperFx.Events.Subscriptions;
 using Fisher;
 using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.DependencyInjection.Extensions;
+using Microsoft.Extensions.Hosting;
 using Microsoft.Extensions.Logging;
 using Weasel.Core;
 using Wolverine.Fisher.Distribution;
@@ -97,12 +99,22 @@ public static FisherStoreConfigurationExpression<T> IntegrateWithWolverine<T>(
             return BuildSingleSqliteMessageStore<T>(schemaName, integration.AutoCreate, store, runtime, logger);
         });
 
-        // Always Fisher's own coordinator, never a Wolverine-managed distributed one: distributing
-        // subscriptions across nodes presupposes several nodes sharing one event store, and an
-        // ancillary Fisher store is a file like any other. GH-3907.
+        // Fisher's own coordinator unless the main integration opts into Wolverine-managed subscription
+        // distribution (GH-5094, mirroring Polecat). That never spreads an ancillary store's projections
+        // across nodes either -- the store is a file like any other (GH-3907) -- but gives its projections
+        // the same agent-based operator surface as the main store's.
+        expression.Services.TryAddSingleton<EventSubscriptionAgentFamily>();
         expression.Services.AddSingleton<IProjectionCoordinator<T>>(s =>
-            new ProjectionCoordinator<T>(s.GetRequiredService<T>(),
-                s.GetRequiredService<ILogger<IProjectionCoordinator>>()));
+        {
+            var store = s.GetRequiredService<T>();
+            if (s.GetService<FisherIntegration>() is { UseWolverineManagedEventSubscriptionDistribution: true })
+            {
+                return new WolverineProjectionCoordinator<T>(s.GetRequiredService<EventSubscriptionAgentFamily>(), store);
+            }
+
+            return new ProjectionCoordinator<T>(store, s.GetRequiredService<ILogger<IProjectionCoordinator>>());
+        });
+        expression.Services.AddSingleton<IHostedService, ManagedDistributionDaemonModeValidator<T>>();
 
         expression.Services.AddSingleton<OutboxedSessionFactory<T>>();
 
@@ -299,8 +311,7 @@ public void Configure(IServiceProvider services, StoreOptions options)
         // messages.
         options.Events.MessageOutbox = new FisherToWolverineOutbox(services);
 
-        // No DaemonSettings.AsyncMode nudge, unlike the Marten and Polecat twins. Those flip to
-        // ExternallyManaged when Wolverine takes over hosting the daemon; Wolverine.Fisher never
-        // does, so the caller's own AddAsyncDaemon() choice for this ancillary store is left alone.
+        // No DaemonSettings.AsyncMode nudge, unlike the Marten and Polecat twins: Fisher has no such
+        // setting, and the caller's own AddAsyncDaemon() choice for this ancillary store is left alone.
     }
 }

src/Persistence/Wolverine.Fisher/Distribution/ManagedDistributionDaemonModeValidator.cs

@@ -0,0 +1,66 @@
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Hosting;
+
+namespace Wolverine.Fisher.Distribution;
+
+/// <summary>
+///     GH-5094, the Fisher twin of Wolverine.Marten's GH-3388 check: refuses a host that combines
+///     Wolverine-managed event subscription distribution with Fisher's own <c>AddAsyncDaemon(DaemonMode.Solo)</c>
+///     on the main store. Fisher's Solo mode hosts its own daemon as JasperFx's
+///     <see cref="JasperFx.Events.Daemon.IProjectionCoordinator" />, and Wolverine's agents would run the same
+///     projections against the same SQLite file beside it. Checked at host start because <c>AddAsyncDaemon</c>
+///     may be called before or after <c>IntegrateWithWolverine</c>.
+/// </summary>
+internal class ManagedDistributionDaemonModeValidator : IHostedService
+{
+    private readonly IServiceProvider _services;
+
+    public ManagedDistributionDaemonModeValidator(IServiceProvider services)
+    {
+        _services = services;
+    }
+
+    public Task StartAsync(CancellationToken cancellationToken)
+    {
+        if (_services.GetServices<JasperFx.Events.Daemon.IProjectionCoordinator>().Any())
+        {
+            throw new InvalidOperationException(Message("the Fisher store"));
+        }
+
+        return Task.CompletedTask;
+    }
+
+    public Task StopAsync(CancellationToken cancellationToken) => Task.CompletedTask;
+
+    internal static string Message(string store) =>
+        $"Fisher's async daemon for {store} is hosted by AddAsyncDaemon(DaemonMode.Solo), but this host also uses Wolverine-managed event subscription distribution (UseWolverineManagedEventSubscriptionDistribution = true). Wolverine runs the projections and subscriptions as agents, so the two would project the same events into the same SQLite file. Remove AddAsyncDaemon() (or use DaemonMode.ExternallyManaged) and let Wolverine's distribution run the daemon, or turn UseWolverineManagedEventSubscriptionDistribution off.";
+}
+
+/// <summary>
+///     GH-5094: the same check for an ancillary Fisher store, whose Solo daemon is registered as JasperFx's
+///     <see cref="JasperFx.Events.Daemon.IProjectionCoordinator{T}" />. The flag lives on the main integration,
+///     so it is read at host start too.
+/// </summary>
+internal class ManagedDistributionDaemonModeValidator<T> : IHostedService where T : class
+{
+    private readonly IServiceProvider _services;
+
+    public ManagedDistributionDaemonModeValidator(IServiceProvider services)
+    {
+        _services = services;
+    }
+
+    public Task StartAsync(CancellationToken cancellationToken)
+    {
+        if (_services.GetService<FisherIntegration>() is { UseWolverineManagedEventSubscriptionDistribution: true }
+            && _services.GetServices<JasperFx.Events.Daemon.IProjectionCoordinator<T>>().Any())
+        {
+            throw new InvalidOperationException(
+                ManagedDistributionDaemonModeValidator.Message($"the ancillary store {typeof(T).Name}"));
+        }
+
+        return Task.CompletedTask;
+    }
+
+    public Task StopAsync(CancellationToken cancellationToken) => Task.CompletedTask;
+}

src/Persistence/Wolverine.Fisher/Distribution/WolverineProjectionCoordinator.cs

@@ -0,0 +1,72 @@
+using JasperFx.Core.Reflection;
+using JasperFx.Events;
+using JasperFx.Events.Daemon;
+using Fisher;
+using Wolverine.Runtime.Agents;
+
+namespace Wolverine.Fisher.Distribution;
+
+/// <summary>
+///     GH-5094: the projection coordinator of a Fisher store under
+///     <see cref="FisherIntegration.UseWolverineManagedEventSubscriptionDistribution" />. Wolverine's
+///     <see cref="EventSubscriptionAgentFamily" /> runs each projection and subscription as an agent, so
+///     operator tooling pauses, restarts and rewinds them the same way on Fisher as on Marten and Polecat.
+/// </summary>
+internal class WolverineProjectionCoordinator<T> : WolverineProjectionCoordinator, IProjectionCoordinator<T>
+    where T : IDocumentStore
+{
+    public WolverineProjectionCoordinator(EventSubscriptionAgentFamily agents, T store) : base(agents, store)
+    {
+    }
+}
+
+internal class WolverineProjectionCoordinator : IProjectionCoordinator
+{
+    private readonly EventStoreAgents _storeAgents;
+
+    public WolverineProjectionCoordinator(EventSubscriptionAgentFamily agents, IDocumentStore store)
+    {
+        _storeAgents = agents.FindStore(store.As<IEventStore>().Identity);
+    }
+
+    public Task StartAsync(CancellationToken cancellationToken)
+    {
+        return _storeAgents.StartAllAsync(cancellationToken);
+    }
+
+    public Task StopAsync(CancellationToken cancellationToken)
+    {
+        return _storeAgents.StopAllAsync(cancellationToken);
+    }
+
+    public ValueTask<IProjectionDaemon> DaemonForMainDatabaseAsync()
+    {
+        return _storeAgents.DaemonForMainDatabaseAsync();
+    }
+
+    // GH-5092: the JasperFx IProjectionCoordinator member, now that this interface extends it
+    public IProjectionDaemon DaemonForMainDatabase()
+    {
+        return _storeAgents.DaemonForMainDatabase();
+    }
+
+    public ValueTask<IProjectionDaemon> DaemonForDatabase(string databaseIdentifier)
+    {
+        return _storeAgents.DaemonForDatabase(databaseIdentifier);
+    }
+
+    public ValueTask<IReadOnlyList<IProjectionDaemon>> AllDaemonsAsync()
+    {
+        return _storeAgents.AllDaemonsAsync();
+    }
+
+    public Task PauseAsync()
+    {
+        return StopAsync(CancellationToken.None);
+    }
+
+    public Task ResumeAsync()
+    {
+        return StartAsync(CancellationToken.None);
+    }
+}

src/Persistence/Wolverine.Fisher/FisherIntegration.cs

@@ -34,6 +34,18 @@ public class FisherIntegration : IWolverineExtension, IEventForwarding
     /// </summary>
     public bool UseFastEventForwarding { get; set; }
 
+    /// <summary>
+    ///     Run this store's async projections and subscriptions as Wolverine agents
+    ///     (<see cref="Wolverine.Runtime.Agents.EventSubscriptionAgentFamily" />), the way
+    ///     <c>UseWolverineManagedEventSubscriptionDistribution</c> does for Marten and Polecat. A Fisher store is
+    ///     one SQLite file, so this does not spread projections over several nodes. What it buys is one operator
+    ///     surface across the three stores: tooling such as CritterWatch pauses, restarts, rebuilds and rewinds a
+    ///     Fisher projection through the agent family exactly as it does a Marten one. Replaces Fisher's own
+    ///     <c>AddAsyncDaemon(DaemonMode.Solo)</c>, which the host refuses alongside it. Applies to ancillary
+    ///     Fisher stores as well. Default is false.
+    /// </summary>
+    public bool UseWolverineManagedEventSubscriptionDistribution { get; set; }
+
 
     public void Configure(WolverineOptions options)
     {
@@ -223,10 +235,9 @@ public void Configure(IServiceProvider services, StoreOptions options)
         options.Events.MessageOutbox = new FisherToWolverineOutbox(services);
 
         // The Marten and Polecat twins nudge DaemonSettings.AsyncMode to ExternallyManaged when
-        // Wolverine takes over hosting the daemon. Wolverine.Fisher never does - see the note in
-        // WolverineOptionsFisherExtensions on why subscription distribution has no meaning over a
-        // single SQLite file - so Fisher's own AddAsyncDaemon() choice is left exactly as the caller
-        // made it.
+        // Wolverine takes over hosting the daemon. Fisher has no such setting: AddAsyncDaemon() registers
+        // its hosted daemon or nothing, so under UseWolverineManagedEventSubscriptionDistribution (GH-5094)
+        // ManagedDistributionDaemonModeValidator refuses an AddAsyncDaemon(DaemonMode.Solo) at host start.
     }
 }