Adopt JasperFx.Events 2.36.2: clear resolved daemons on coordinator stop, idempotent AddAsyncDaemon

Directory.Packages.props

@@ -152,14 +152,20 @@
          the lost-wakeup window between those two paths but left it open, so a watcher could fall between
          them and see the state on NEITHER — and once a high water agent reaches the head it has nothing
          left to publish, so the wait could only end in a timeout no matter how generous. That surfaced
-         as HighWaterAgentTests.skips_multiple_gaps_and_keeps_advancing failing ~8 of 10 runs. -->
-    <PackageVersion Include="JasperFx" Version="2.36.1" />
-    <PackageVersion Include="JasperFx.Events" Version="2.36.1" />
-    <PackageVersion Include="JasperFx.Events.SourceGenerator" Version="2.36.1">
+         as HighWaterAgentTests.skips_multiple_gaps_and_keeps_advancing failing ~8 of 10 runs.
+         JasperFx 2.36.2: jasperfx#574/#575 (marten#5055/#5056) — the daemon stop path is now safe to
+         run after disposal: JasperFxAsyncDaemon.Dispose() is idempotent and StopAllAsync() no-ops on a
+         disposed daemon, ProjectionCoordinatorBase.PauseAsync logs ObjectDisposedException at Debug
+         instead of Error, and ProjectionCoordinatorBase.StopAsync calls the new abstract
+         ClearResolvedDaemons() seam after disposing daemons so subclass caches drop the disposed
+         instances (implemented here on ProjectionCoordinator + ExplicitProjectionCoordinator). -->
+    <PackageVersion Include="JasperFx" Version="2.36.2" />
+    <PackageVersion Include="JasperFx.Events" Version="2.36.2" />
+    <PackageVersion Include="JasperFx.Events.SourceGenerator" Version="2.36.2">
       <PrivateAssets>all</PrivateAssets>
       <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
     </PackageVersion>
-    <PackageVersion Include="JasperFx.SourceGenerator" Version="2.36.1" />
+    <PackageVersion Include="JasperFx.SourceGenerator" Version="2.36.2" />
     <PackageVersion Include="Jil" Version="3.0.0-alpha2" />
     <PackageVersion Include="Lamar" Version="7.1.1" />
     <PackageVersion Include="Lamar.Microsoft.DependencyInjection" Version="15.0.0" />

src/DaemonTests/Bug_5055_5056_coordinator_double_stop.cs

@@ -0,0 +1,215 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Threading;
+using System.Threading.Tasks;
+using DaemonTests.TestingSupport;
+using JasperFx;
+using JasperFx.Core;
+using JasperFx.Events.Daemon;
+using JasperFx.Events.Projections;
+using Marten;
+using Marten.Events.Aggregation;
+using Marten.Events.Daemon.Coordination;
+using Marten.Testing.Harness;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Hosting;
+using Microsoft.Extensions.Logging;
+using Shouldly;
+using Xunit;
+using Xunit.Abstractions;
+
+namespace DaemonTests;
+
+public record Bug5055Event();
+
+public class Bug5055Stream { public Guid Id { get; set; } }
+
+public partial class Bug5055Projection: SingleStreamProjection<Bug5055Stream, Guid>
+{
+    public void Apply(Bug5055Event @event, Bug5055Stream projection) { }
+}
+
+public interface IBug5056Store: IDocumentStore;
+
+// #5055/#5056 — at pod shutdown a second Pause/Stop pass over the coordinator fanned StopAllAsync out
+// over daemons the first pass had already disposed, logging one ObjectDisposedException-backed
+// "Error while trying to stop daemon agents" per daemon. The jasperfx#574 fix (2.36.2) makes the
+// daemon stop path safe after disposal and adds the ClearResolvedDaemons() seam; the Marten side
+// (#5056) clears the daemon cache on StopAsync and makes AddAsyncDaemon registration idempotent so
+// the host can't end up stopping the same coordinator twice.
+public class Bug_5055_5056_coordinator_double_stop: DaemonContext
+{
+    public Bug_5055_5056_coordinator_double_stop(ITestOutputHelper output): base(output)
+    {
+    }
+
+    private const string Shard = "Bug5055Stream:All";
+
+    [Fact]
+    public async Task stopping_the_coordinator_twice_is_quiet_and_safe()
+    {
+        var logger = new CapturingLogger(_output);
+
+        StoreOptions(x =>
+        {
+            x.Projections.Add(new Bug5055Projection(), ProjectionLifecycle.Async);
+            x.Projections.AsyncMode = DaemonMode.Solo;
+        });
+
+        var coordinator = new ProjectionCoordinator(theStore, logger);
+        await coordinator.StartAsync(CancellationToken.None);
+
+        try
+        {
+            // Get the coordinator genuinely running agents before shutting down
+            await using (var session = theStore.LightweightSession())
+            {
+                for (var i = 0; i < 10; i++)
+                {
+                    session.Events.Append(Guid.NewGuid(), new Bug5055Event());
+                }
+
+                await session.SaveChangesAsync();
+            }
+
+            var daemon = coordinator.DaemonForMainDatabase();
+            await daemon.Tracker.WaitForShardState(new ShardState(Shard, 10), 30.Seconds());
+        }
+        finally
+        {
+            await coordinator.StopAsync(CancellationToken.None);
+        }
+
+        // The second pass is the #5055 shutdown shape (double hosted-service registration, user
+        // pause + host stop, Wolverine quiesce + host stop). It must find nothing to stop.
+        await coordinator.StopAsync(CancellationToken.None);
+
+        logger.Entries.Where(x => x.Level == LogLevel.Error)
+            .ShouldNotContain(x => x.Message.Contains("stop daemon agents"));
+        logger.Entries.ShouldNotContain(x => x.Exception is ObjectDisposedException && x.Level == LogLevel.Error);
+    }
+
+    [Fact]
+    public async Task daemon_accessors_after_stop_hand_back_fresh_daemons()
+    {
+        var logger = new CapturingLogger(_output);
+
+        StoreOptions(x =>
+        {
+            x.Projections.Add(new Bug5055Projection(), ProjectionLifecycle.Async);
+            x.Projections.AsyncMode = DaemonMode.Solo;
+        });
+
+        var coordinator = new ProjectionCoordinator(theStore, logger);
+
+        var before = coordinator.DaemonForMainDatabase();
+        await coordinator.StopAsync(CancellationToken.None);
+
+        // jasperfx#574: the daemon StopAsync just disposed no-ops instead of throwing
+        // ObjectDisposedException off its disposed CancellationTokenSource
+        await before.StopAllAsync();
+
+        // marten#5056: the cache was cleared, so the accessor rebuilds a fresh, usable daemon
+        // instead of handing back the disposed instance
+        var after = coordinator.DaemonForMainDatabase();
+        ReferenceEquals(before, after).ShouldBeFalse();
+
+        try
+        {
+            await after.StartAllAsync();
+            await after.StopAllAsync();
+        }
+        finally
+        {
+            after.Dispose();
+        }
+    }
+
+    [Fact]
+    public void add_async_daemon_twice_registers_a_single_hosted_coordinator()
+    {
+        var once = registerMarten(1);
+        var twice = registerMarten(2);
+
+        twice.Count(x => x.ServiceType == typeof(Marten.Events.Daemon.Coordination.IProjectionCoordinator))
+            .ShouldBe(1);
+        twice.Count(x => x.ServiceType == typeof(IHostedService))
+            .ShouldBe(once.Count(x => x.ServiceType == typeof(IHostedService)));
+    }
+
+    [Fact]
+    public void add_async_daemon_twice_on_a_separate_store_registers_a_single_hosted_coordinator()
+    {
+        var once = registerMartenStore(1);
+        var twice = registerMartenStore(2);
+
+        twice.Count(x => x.ServiceType == typeof(Marten.Events.Daemon.Coordination.IProjectionCoordinator<IBug5056Store>))
+            .ShouldBe(1);
+        twice.Count(x => x.ServiceType == typeof(IHostedService))
+            .ShouldBe(once.Count(x => x.ServiceType == typeof(IHostedService)));
+    }
+
+    private static IServiceCollection registerMarten(int addAsyncDaemonCalls)
+    {
+        var services = new ServiceCollection();
+        var expression = services.AddMarten(opts => opts.Connection(ConnectionSource.ConnectionString));
+        for (var i = 0; i < addAsyncDaemonCalls; i++)
+        {
+            expression.AddAsyncDaemon(DaemonMode.HotCold);
+        }
+
+        return services;
+    }
+
+    private static IServiceCollection registerMartenStore(int addAsyncDaemonCalls)
+    {
+        var services = new ServiceCollection();
+        var expression = services.AddMartenStore<IBug5056Store>(opts =>
+            opts.Connection(ConnectionSource.ConnectionString));
+        for (var i = 0; i < addAsyncDaemonCalls; i++)
+        {
+            expression.AddAsyncDaemon(DaemonMode.HotCold);
+        }
+
+        return services;
+    }
+
+    // Records every log entry so the double-stop pass can be asserted quiet; TestLogger<T> only
+    // echoes to output. Thread-safe: the coordinator loop and the test body log concurrently.
+    private sealed class CapturingLogger: ILogger<ProjectionCoordinator>
+    {
+        private readonly ITestOutputHelper _output;
+        private readonly object _gate = new();
+
+        public CapturingLogger(ITestOutputHelper output) => _output = output;
+
+        public List<(LogLevel Level, string Message, Exception? Exception)> Entries { get; } = new();
+
+        public IDisposable BeginScope<TState>(TState state) where TState : notnull => NullScope.Instance;
+
+        public bool IsEnabled(LogLevel logLevel) => true;
+
+        public void Log<TState>(LogLevel logLevel, EventId eventId, TState state, Exception? exception,
+            Func<TState, Exception?, string> formatter)
+        {
+            var message = formatter(state, exception);
+            lock (_gate)
+            {
+                Entries.Add((logLevel, message, exception));
+            }
+
+            _output.WriteLine($"{logLevel}: {message}");
+            if (exception != null)
+            {
+                _output.WriteLine(exception.ToString());
+            }
+        }
+
+        private sealed class NullScope: IDisposable
+        {
+            public static readonly NullScope Instance = new();
+            public void Dispose() { }
+        }
+    }
+}

src/Marten/Events/Daemon/Coordination/ExplicitProjectionCoordinator.cs

@@ -100,6 +100,16 @@ public async Task StopAsync(CancellationToken cancellationToken)
             {
                 _logger.LogError(exception, "Error while trying to stop daemon agents in database {Name}", pair.Key);
             }
+
+            pair.Value.SafeDispose();
+        }
+
+        // marten#5056 (jasperfx#574): same posture as ProjectionCoordinatorBase.StopAsync. The daemons
+        // above were just disposed, so drop them from the cache — a repeated Pause/Stop has nothing to
+        // fan out over, and the external manager re-acquires fresh daemons through the accessors.
+        lock (_daemonLock)
+        {
+            _daemons = ImHashMap<string, IProjectionDaemon>.Empty;
         }
     }
 

src/Marten/Events/Daemon/Coordination/ProjectionCoordinator.cs

@@ -148,6 +148,17 @@ protected override IReadOnlyList<IProjectionDaemon> ResolvedDaemons()
         return _daemons.Enumerate().Select(x => x.Value).ToList();
     }
 
+    // marten#5056 (jasperfx#574): StopAsync just disposed every resolved daemon. Drop them from the
+    // cache so a repeated Pause/Stop has nothing to fan out over and a later ResumeAsync or daemon
+    // accessor builds fresh daemons instead of handing back disposed instances.
+    protected override void ClearResolvedDaemons()
+    {
+        lock (_daemonLock)
+        {
+            _daemons = ImHashMap<string, IProjectionDaemon>.Empty;
+        }
+    }
+
     public override IProjectionDaemon DaemonForMainDatabase()
     {
         var database = (MartenDatabase)Store.Tenancy.Default.Database;

src/Marten/MartenServiceCollectionExtensions.cs

@@ -516,7 +516,13 @@ public MartenStoreExpression<T> AddAsyncDaemon(DaemonMode mode)
             // ExternallyManaged (jasperfx#490) means an external system (e.g. Wolverine's managed
             // event-subscription distribution) executes the async projections — Marten must not
             // register its own coordination for either.
-            if (mode is DaemonMode.Solo or DaemonMode.HotCold)
+            // marten#5056: guard against a repeated AddAsyncDaemon() call registering a second
+            // IHostedService forwarding to the same coordinator singleton. The host would call
+            // StopAsync twice on it at shutdown, and the second pass used to fan StopAllAsync out
+            // over daemons the first pass had already disposed (the marten#5055 error-log storm).
+            if (mode is DaemonMode.Solo or DaemonMode.HotCold && !Services.Any(x =>
+                    x.ServiceType == typeof(Marten.Events.Daemon.Coordination.IProjectionCoordinator<T>) &&
+                    x.ImplementationType == typeof(ProjectionCoordinator<T>)))
             {
                 Services.AddSingleton<Marten.Events.Daemon.Coordination.IProjectionCoordinator<T>, ProjectionCoordinator<T>>();
                 Services.AddSingleton<IHostedService>(s => s.GetRequiredService<Marten.Events.Daemon.Coordination.IProjectionCoordinator<T>>());
@@ -763,7 +769,13 @@ public MartenConfigurationExpression AddAsyncDaemon(DaemonMode mode)
             // ExternallyManaged (jasperfx#490) means an external system (e.g. Wolverine's managed
             // event-subscription distribution) executes the async projections — Marten must not
             // register its own coordination for either.
-            if (mode is DaemonMode.Solo or DaemonMode.HotCold)
+            // marten#5056: guard against a repeated AddAsyncDaemon() call registering a second
+            // IHostedService forwarding to the same coordinator singleton. The host would call
+            // StopAsync twice on it at shutdown, and the second pass used to fan StopAllAsync out
+            // over daemons the first pass had already disposed (the marten#5055 error-log storm).
+            if (mode is DaemonMode.Solo or DaemonMode.HotCold && !Services.Any(x =>
+                    x.ServiceType == typeof(Marten.Events.Daemon.Coordination.IProjectionCoordinator) &&
+                    x.ImplementationType == typeof(ProjectionCoordinator)))
             {
                 Services.AddSingleton<Marten.Events.Daemon.Coordination.IProjectionCoordinator, ProjectionCoordinator>();
                 Services.AddSingleton<IHostedService>(s => s.GetRequiredService<Marten.Events.Daemon.Coordination.IProjectionCoordinator>());