From 1186c662ac2fc36922fc2781e0265fc55de4c5a1 Mon Sep 17 00:00:00 2001 From: "Jeremy D. Miller" Date: Tue, 20 Oct 2026 08:28:35 -0500 Subject: [PATCH] GH-5094 Wolverine.Fisher: opt-in Wolverine-managed event subscription distribution FisherIntegration.UseWolverineManagedEventSubscriptionDistribution (default false) runs a Fisher store's async projections and subscriptions as Wolverine event subscription agents, as the Marten and Polecat integrations do. A Fisher store is one SQLite file, so nothing spreads across nodes. The point is one operator surface: CritterWatch pauses, restarts, rebuilds and rewinds projections through IEventSubscriptionAgentFamily, and a Fisher host had no family at all (CritterWatch #1437). - Main store: registers EventSubscriptionAgentFamily as IAgentFamily and IEventSubscriptionAgentFamily, and a WolverineProjectionCoordinator as Wolverine.Fisher's IProjectionCoordinator, so the Fisher testing helpers drive the agents. - Ancillary stores: IProjectionCoordinator becomes the Wolverine-managed one when the main integration opts in, mirroring Wolverine.Polecat. - ManagedDistributionDaemonModeValidator (main and per ancillary store) refuses Fisher's own AddAsyncDaemon(DaemonMode.Solo) alongside it at host start, the Fisher twin of Marten's GH-3388 check: both would project the same events into the same file. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01DCggXgxnG3EBXoR9rrq37U --- ...ed_event_subscription_distribution_5094.cs | 171 ++++++++++++++++++ ...cillaryWolverineOptionsFisherExtensions.cs | 27 ++- .../ManagedDistributionDaemonModeValidator.cs | 66 +++++++ .../WolverineProjectionCoordinator.cs | 72 ++++++++ .../Wolverine.Fisher/FisherIntegration.cs | 19 +- .../WolverineOptionsFisherExtensions.cs | 24 ++- 6 files changed, 363 insertions(+), 16 deletions(-) create mode 100644 src/Persistence/FisherTests/fisher_managed_event_subscription_distribution_5094.cs create mode 100644 src/Persistence/Wolverine.Fisher/Distribution/ManagedDistributionDaemonModeValidator.cs create mode 100644 src/Persistence/Wolverine.Fisher/Distribution/WolverineProjectionCoordinator.cs diff --git a/src/Persistence/FisherTests/fisher_managed_event_subscription_distribution_5094.cs b/src/Persistence/FisherTests/fisher_managed_event_subscription_distribution_5094.cs new file mode 100644 index 0000000000..69f2258691 --- /dev/null +++ b/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; + +/// +/// GH-5094. With , 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. +/// +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(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().ShouldHaveSingleItem(); + _host.Services.GetServices().OfType().ShouldHaveSingleItem(); + + // Wolverine runs the daemon, not Fisher's own hosted service + _host.Services.GetServices().ShouldBeEmpty(); + _host.Services.GetRequiredService().GetType().Name.ShouldBe("WolverineProjectionCoordinator"); + } + + [Fact] + public async Task the_projection_is_an_agent_that_tooling_can_find() + { + var family = _host.Services.GetRequiredService(); + + 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(); + 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(); + await using (var session = store.LightweightSession()) + { + session.Events.StartStream(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(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(); + 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(); + 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(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(SnapshotLifecycle.Async); + }) + .AddAsyncDaemon(DaemonMode.Solo) + .IntegrateWithWolverine(x => x.UseWolverineManagedEventSubscriptionDistribution = true); + }).StartAsync(); + }); + + ex.Message.ShouldContain("UseWolverineManagedEventSubscriptionDistribution"); + } + + private static Task waitFor(Func condition, string what) => waitFor(() => Task.FromResult(condition()), what); + + private static async Task waitFor(Func> 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; +} diff --git a/src/Persistence/Wolverine.Fisher/AncillaryWolverineOptionsFisherExtensions.cs b/src/Persistence/Wolverine.Fisher/AncillaryWolverineOptionsFisherExtensions.cs index 8fddcd7e80..70b50364ba 100644 --- a/src/Persistence/Wolverine.Fisher/AncillaryWolverineOptionsFisherExtensions.cs +++ b/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 IntegrateWithWolverine( return BuildSingleSqliteMessageStore(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(); expression.Services.AddSingleton>(s => - new ProjectionCoordinator(s.GetRequiredService(), - s.GetRequiredService>())); + { + var store = s.GetRequiredService(); + if (s.GetService() is { UseWolverineManagedEventSubscriptionDistribution: true }) + { + return new WolverineProjectionCoordinator(s.GetRequiredService(), store); + } + + return new ProjectionCoordinator(store, s.GetRequiredService>()); + }); + expression.Services.AddSingleton>(); expression.Services.AddSingleton>(); @@ -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. } } diff --git a/src/Persistence/Wolverine.Fisher/Distribution/ManagedDistributionDaemonModeValidator.cs b/src/Persistence/Wolverine.Fisher/Distribution/ManagedDistributionDaemonModeValidator.cs new file mode 100644 index 0000000000..06270634bb --- /dev/null +++ b/src/Persistence/Wolverine.Fisher/Distribution/ManagedDistributionDaemonModeValidator.cs @@ -0,0 +1,66 @@ +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Hosting; + +namespace Wolverine.Fisher.Distribution; + +/// +/// 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 AddAsyncDaemon(DaemonMode.Solo) +/// on the main store. Fisher's Solo mode hosts its own daemon as JasperFx's +/// , and Wolverine's agents would run the same +/// projections against the same SQLite file beside it. Checked at host start because AddAsyncDaemon +/// may be called before or after IntegrateWithWolverine. +/// +internal class ManagedDistributionDaemonModeValidator : IHostedService +{ + private readonly IServiceProvider _services; + + public ManagedDistributionDaemonModeValidator(IServiceProvider services) + { + _services = services; + } + + public Task StartAsync(CancellationToken cancellationToken) + { + if (_services.GetServices().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."; +} + +/// +/// GH-5094: the same check for an ancillary Fisher store, whose Solo daemon is registered as JasperFx's +/// . The flag lives on the main integration, +/// so it is read at host start too. +/// +internal class ManagedDistributionDaemonModeValidator : IHostedService where T : class +{ + private readonly IServiceProvider _services; + + public ManagedDistributionDaemonModeValidator(IServiceProvider services) + { + _services = services; + } + + public Task StartAsync(CancellationToken cancellationToken) + { + if (_services.GetService() is { UseWolverineManagedEventSubscriptionDistribution: true } + && _services.GetServices>().Any()) + { + throw new InvalidOperationException( + ManagedDistributionDaemonModeValidator.Message($"the ancillary store {typeof(T).Name}")); + } + + return Task.CompletedTask; + } + + public Task StopAsync(CancellationToken cancellationToken) => Task.CompletedTask; +} diff --git a/src/Persistence/Wolverine.Fisher/Distribution/WolverineProjectionCoordinator.cs b/src/Persistence/Wolverine.Fisher/Distribution/WolverineProjectionCoordinator.cs new file mode 100644 index 0000000000..a4bfb7c501 --- /dev/null +++ b/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; + +/// +/// GH-5094: the projection coordinator of a Fisher store under +/// . Wolverine's +/// 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. +/// +internal class WolverineProjectionCoordinator : WolverineProjectionCoordinator, IProjectionCoordinator + 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().Identity); + } + + public Task StartAsync(CancellationToken cancellationToken) + { + return _storeAgents.StartAllAsync(cancellationToken); + } + + public Task StopAsync(CancellationToken cancellationToken) + { + return _storeAgents.StopAllAsync(cancellationToken); + } + + public ValueTask DaemonForMainDatabaseAsync() + { + return _storeAgents.DaemonForMainDatabaseAsync(); + } + + // GH-5092: the JasperFx IProjectionCoordinator member, now that this interface extends it + public IProjectionDaemon DaemonForMainDatabase() + { + return _storeAgents.DaemonForMainDatabase(); + } + + public ValueTask DaemonForDatabase(string databaseIdentifier) + { + return _storeAgents.DaemonForDatabase(databaseIdentifier); + } + + public ValueTask> AllDaemonsAsync() + { + return _storeAgents.AllDaemonsAsync(); + } + + public Task PauseAsync() + { + return StopAsync(CancellationToken.None); + } + + public Task ResumeAsync() + { + return StartAsync(CancellationToken.None); + } +} diff --git a/src/Persistence/Wolverine.Fisher/FisherIntegration.cs b/src/Persistence/Wolverine.Fisher/FisherIntegration.cs index b8baed9fc6..2f000f568b 100644 --- a/src/Persistence/Wolverine.Fisher/FisherIntegration.cs +++ b/src/Persistence/Wolverine.Fisher/FisherIntegration.cs @@ -34,6 +34,18 @@ public class FisherIntegration : IWolverineExtension, IEventForwarding /// public bool UseFastEventForwarding { get; set; } + /// + /// Run this store's async projections and subscriptions as Wolverine agents + /// (), the way + /// UseWolverineManagedEventSubscriptionDistribution 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 + /// AddAsyncDaemon(DaemonMode.Solo), which the host refuses alongside it. Applies to ancillary + /// Fisher stores as well. Default is false. + /// + 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. } } diff --git a/src/Persistence/Wolverine.Fisher/WolverineOptionsFisherExtensions.cs b/src/Persistence/Wolverine.Fisher/WolverineOptionsFisherExtensions.cs index 31bec70ddf..8fe3f123af 100644 --- a/src/Persistence/Wolverine.Fisher/WolverineOptionsFisherExtensions.cs +++ b/src/Persistence/Wolverine.Fisher/WolverineOptionsFisherExtensions.cs @@ -9,6 +9,9 @@ using JasperFx.Events.Documents; using Microsoft.Extensions.DependencyInjection.Extensions; using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Hosting; +using Wolverine.Fisher.Distribution; +using Wolverine.Runtime.Agents; using Wolverine.Fisher.Publishing; using Wolverine.Fisher.Subscriptions; using Wolverine.Persistence; @@ -129,10 +132,23 @@ public static FisherConfigurationExpression IntegrateWithWolverine( // integrations each learned this the expensive way (GH-3133 / GH-3219); Fisher inherits the // conclusion rather than the detour. - // No Wolverine-managed subscription distribution here, unlike the Marten and Polecat twins. - // Distributing subscriptions across nodes presupposes several nodes sharing one event store, - // and a Fisher store is a single SQLite file - Fisher itself refuses DaemonMode.HotCold for - // the same reason. Run the daemon in Solo mode via Fisher's own AddAsyncDaemon(). + // GH-5094: Wolverine-managed subscription distribution is opt-in. A Fisher store is a single SQLite + // file, so there is nothing to spread across nodes (Fisher itself refuses DaemonMode.HotCold for the + // same reason), and Fisher's own AddAsyncDaemon(DaemonMode.Solo) stays the default. The agent family + // gives operator tooling the same pause / restart / rebuild / rewind surface it has on Marten and + // Polecat. + if (integration.UseWolverineManagedEventSubscriptionDistribution) + { + expression.Services.AddSingleton(); + expression.Services.TryAddSingleton(); + expression.Services.AddSingleton(s => s.GetRequiredService()); + expression.Services.AddSingleton(s => + s.GetRequiredService()); + expression.Services.AddSingleton(s => + s.GetRequiredService()); + + expression.Services.AddSingleton(); + } return expression; }