fix(#574): stop fanning StopAllAsync out over disposed daemons at shutdown (marten#5055)

src/EventTests/Daemon/DisposedDaemonStopAllTests.cs

@@ -0,0 +1,88 @@
+using System.Diagnostics.Metrics;
+using EventTests.Projections;
+using JasperFx.Events;
+using JasperFx.Events.Daemon;
+using JasperFx.Events.Daemon.HighWater;
+using JasperFx.Events.Projections;
+using NSubstitute;
+using Shouldly;
+
+namespace EventTests.Daemon;
+
+// marten#5055: at Kubernetes pod shutdown a second Pause/Stop pass (double AddAsyncDaemon
+// hosted-service registration, user pause + host stop, Wolverine quiesce + host stop) fans
+// StopAllAsync out over daemons the first pass already disposed. StopAllAsync used to open with
+// _semaphore.WaitAsync(_cancellation.Token), and reading .Token off the disposed source threw
+// ObjectDisposedException — one "Error while trying to stop daemon agents" log per daemon, on
+// every shutdown. A disposed daemon has nothing left to stop, so StopAllAsync must be a no-op.
+public class DisposedDaemonStopAllTests
+{
+    private static JasperFxAsyncDaemon<FakeOperations, FakeSession, IJasperFxProjection<FakeOperations>> BuildDaemon()
+    {
+        var store = Substitute.For<IEventStore<FakeOperations, FakeSession>>();
+        store.Meter.Returns(new Meter("tests"));
+        store.TimeProvider.Returns(TimeProvider.System);
+
+        var database = Substitute.For<IEventDatabase>();
+        database.Identifier.Returns("db1");
+        database.DatabaseUri.Returns(new Uri("fake://db1"));
+        database.Tracker.Returns(new ShardStateTracker(new NulloLogger()));
+
+        return new JasperFxAsyncDaemon<FakeOperations, FakeSession, IJasperFxProjection<FakeOperations>>(
+            store, database, new NulloLogger(), new StubDetector(), new FakeProjectionGraph());
+    }
+
+    [Fact]
+    public async Task stop_all_after_dispose_is_a_no_op_instead_of_throwing()
+    {
+        var daemon = BuildDaemon();
+        daemon.Dispose();
+
+        await Should.NotThrowAsync(daemon.StopAllAsync);
+    }
+
+    [Fact]
+    public async Task stop_all_after_a_stop_dispose_cycle_is_still_a_no_op()
+    {
+        // The exact shutdown shape from the issue: the first coordinator pass stops then disposes,
+        // the second pass calls StopAllAsync again on the same instance.
+        var daemon = BuildDaemon();
+        await daemon.StopAllAsync();
+        daemon.Dispose();
+
+        await Should.NotThrowAsync(daemon.StopAllAsync);
+    }
+
+    [Fact]
+    public void dispose_is_idempotent()
+    {
+        var daemon = BuildDaemon();
+        daemon.Dispose();
+
+        Should.NotThrow(daemon.Dispose);
+    }
+
+    private sealed class StubDetector : IHighWaterDetector
+    {
+        public Uri DatabaseUri { get; } = new("fake://db1");
+
+        public Task<HighWaterStatistics> Detect(CancellationToken token)
+            => Task.FromResult(new HighWaterStatistics());
+
+        public Task<HighWaterStatistics> DetectInSafeZone(CancellationToken token) => Detect(token);
+    }
+
+    // Minimal concrete ProjectionGraph — on these paths the daemon consumes it only as DaemonSettings.
+    private sealed class FakeProjectionGraph :
+        ProjectionGraph<IJasperFxProjection<FakeOperations>, FakeOperations, FakeSession>
+    {
+        public FakeProjectionGraph() : base(Substitute.For<IEventRegistry>(), "tests")
+        {
+        }
+
+        protected override void onAddProjection(object projection)
+        {
+            // Nothing
+        }
+    }
+}

src/EventTests/Daemon/ProjectionCoordinatorBaseTests.cs

@@ -289,6 +289,74 @@ public async Task stop_releases_all_locks_and_disposes_daemons()
         daemon.StopAllCount.ShouldBeGreaterThanOrEqualTo(1);
     }
 
+    // ---- marten#5055: shutdown over already-disposed daemons ----
+    //
+    // StopAsync disposes every resolved daemon, but the subclass cache used to keep handing them
+    // back. Any second Pause/Stop (double AddAsyncDaemon hosted-service registration, user pause +
+    // host stop, Wolverine quiesce + host stop) then fanned StopAllAsync out over disposed daemons,
+    // logging "Error while trying to stop daemon agents" with an ObjectDisposedException per daemon
+    // on every pod shutdown.
+
+    [Fact]
+    public async Task stop_purges_the_resolved_daemon_cache()
+    {
+        var daemon = new FakeDaemon();
+        var distributor = new FakeDistributor([]);
+
+        var coordinator = BuildCoordinator(distributor, daemon);
+        await coordinator.StartAsync(CancellationToken.None);
+        coordinator.ForceResolve();
+
+        await coordinator.StopAsync(CancellationToken.None);
+
+        // The disposed daemons are gone, so a later ResumeAsync (or daemon accessor path) has to
+        // resolve fresh ones instead of handing back a dead instance.
+        coordinator.VisibleResolvedDaemons.ShouldBeEmpty();
+    }
+
+    [Fact]
+    public async Task a_second_stop_neither_throws_nor_logs_an_error_over_the_disposed_daemons()
+    {
+        var logger = new CapturingLogger();
+        // Behaves like the pre-fix real daemon: StopAllAsync throws ObjectDisposedException once the
+        // daemon has been disposed, so this test fails if the cache purge ever regresses.
+        var daemon = new FakeDaemon { ThrowObjectDisposedOnStopAllWhenDisposed = true };
+        var distributor = new FakeDistributor([]);
+
+        var coordinator = BuildCoordinator(distributor, daemon, logger: logger);
+        await coordinator.StartAsync(CancellationToken.None);
+        coordinator.ForceResolve();
+
+        await coordinator.StopAsync(CancellationToken.None);
+        await Should.NotThrowAsync(() => coordinator.StopAsync(CancellationToken.None));
+
+        // The first stop did the real work; the second had nothing to fan out over.
+        daemon.StopAllCount.ShouldBe(1);
+        daemon.DisposeCount.ShouldBe(1);
+        logger.Errors.ShouldBeEmpty();
+    }
+
+    [Fact]
+    public async Task pausing_over_an_already_disposed_daemon_logs_debug_not_error()
+    {
+        // A daemon disposed out from under the coordinator (e.g. a store shutdown racing a pause)
+        // still surfaces ObjectDisposedException from StopAllAsync. That is benign — there is
+        // nothing left to stop — so it must land at Debug, not as the issue's Error log.
+        var logger = new CapturingLogger();
+        var daemon = new FakeDaemon { ThrowObjectDisposedOnStopAllWhenDisposed = true };
+        var distributor = new FakeDistributor([]);
+
+        var coordinator = BuildCoordinator(distributor, daemon, logger: logger);
+        await coordinator.StartAsync(CancellationToken.None);
+        coordinator.ForceResolve();
+        daemon.Dispose();
+
+        await Should.NotThrowAsync(() => coordinator.PauseAsync());
+
+        logger.Errors.ShouldBeEmpty();
+        logger.Debugs.ShouldContain(x => x.Contains("already disposed"));
+    }
+
     // #352: a coordinator can be legitimately constructed but never started when the async
     // daemon is Disabled — the store's BuildDistributor returns null because there is nothing
     // to coordinate. The ctor must not reject that; the lifecycle methods must no-op cleanly.
@@ -334,11 +402,13 @@ private static async Task WaitFor(Func<bool> condition, int timeoutMs = 3000)
         }
     }
 
-    // Minimal ILogger that records the formatted message of every Error-level entry so a test
-    // can assert whether the loop logged a lock error (proving which catch branch fired).
+    // Minimal ILogger that records the formatted message of every Error- and Debug-level entry so a
+    // test can assert which catch branch fired (e.g. marten#5055's benign-disposal Debug vs. the
+    // "Error while trying to stop daemon agents" Error).
     private sealed class CapturingLogger : ILogger
     {
         public List<string> Errors { get; } = [];
+        public List<string> Debugs { get; } = [];
 
         public void Log<TState>(LogLevel logLevel, EventId eventId, TState state, Exception? exception,
             Func<TState, Exception?, string> formatter)
@@ -350,6 +420,13 @@ public void Log<TState>(LogLevel logLevel, EventId eventId, TState state, Except
                     Errors.Add(formatter(state, exception));
                 }
             }
+            else if (logLevel == LogLevel.Debug)
+            {
+                lock (Debugs)
+                {
+                    Debugs.Add(formatter(state, exception));
+                }
+            }
         }
 
         public bool IsEnabled(LogLevel logLevel) => true;
@@ -388,6 +465,11 @@ protected override IProjectionDaemon ResolveDaemon(IProjectionSet set)
         protected override IReadOnlyList<IProjectionDaemon> ResolvedDaemons()
             => _resolved ? [_daemon] : [];
 
+        protected override void ClearResolvedDaemons() => _resolved = false;
+
+        // Exposes the protected snapshot so marten#5055 tests can assert the cache is purged.
+        public IReadOnlyList<IProjectionDaemon> VisibleResolvedDaemons => ResolvedDaemons();
+
         public override IProjectionDaemon DaemonForMainDatabase() => _daemon;
 
         public override ValueTask<IProjectionDaemon> DaemonForDatabase(string databaseIdentifier)
@@ -505,6 +587,10 @@ private sealed class FakeDaemon : IProjectionDaemon
         public int DisposeCount { get; private set; }
         public int StopAllCount { get; private set; }
 
+        // marten#5055: mimics the real JasperFxAsyncDaemon BEFORE the fix — StopAllAsync on a disposed
+        // daemon read _cancellation.Token off a disposed CancellationTokenSource and threw.
+        public bool ThrowObjectDisposedOnStopAllWhenDisposed { get; init; }
+
         private readonly List<ISubscriptionAgent> _agents = [];
         private bool _failed;
 
@@ -579,6 +665,11 @@ public IReadOnlyList<ISubscriptionAgent> CurrentAgents()
 
         public Task StopAllAsync()
         {
+            if (ThrowObjectDisposedOnStopAllWhenDisposed && DisposeCount > 0)
+            {
+                throw new ObjectDisposedException(nameof(CancellationTokenSource));
+            }
+
             StopAllCount++;
             return Task.CompletedTask;
         }

src/JasperFx.Events/Daemon/JasperFxAsyncDaemon.cs

@@ -41,6 +41,11 @@ public partial class JasperFxAsyncDaemon<TOperations, TQuerySession, TProjection
     private RetryBlock<DeadLetterEvent> _deadLetterBlock;
     private readonly SemaphoreSlim _semaphore = new SemaphoreSlim(1, 1);
 
+    // marten#5055: StopAllAsync on an already-disposed daemon must be a no-op instead of throwing
+    // ObjectDisposedException from _cancellation.Token. Volatile because Dispose() (sync, e.g. from
+    // ProjectionCoordinatorBase.StopAsync) can race a StopAllAsync fanned out from another Pause/Stop.
+    private volatile bool _disposed;
+
     // Only non-null when the backing store partitions events per tenant; null keeps the daemon on the
     // single store-global high-water mark (today's behavior, byte for byte). jasperfx#407 Phase 2b.
     private readonly TenantedHighWaterCoordinator? _tenantHighWater;
@@ -227,6 +232,13 @@ private ExtendedProgressionWriter buildExtendedProgressionWriter()
 
     public void Dispose()
     {
+        // marten#5055: idempotent, and flags StopAllAsync to no-op once the daemon is gone. Double
+        // disposal is a real path: ProjectionCoordinatorBase.StopAsync disposes every resolved daemon,
+        // and a second Pause/Stop (double hosted-service registration, user pause + host stop) fans
+        // back out over the same instances.
+        if (_disposed) return;
+        _disposed = true;
+
         _cancellation?.Dispose();
         _highWater?.Dispose();
         _tenantHighWaterTimer?.Stop();
@@ -888,7 +900,22 @@ private async Task buildPerTenantContinuousShards(
 
     public async Task StopAllAsync()
     {
-        await _semaphore.WaitAsync(_cancellation.Token).ConfigureAwait(false);
+        // marten#5055: a disposed daemon has nothing left to stop. Without this guard, the second
+        // Pause/Stop pass at shutdown (double AddAsyncDaemon registration, user pause + host stop,
+        // Wolverine quiesce + host stop) hits _cancellation.Token on a disposed source and throws
+        // ObjectDisposedException, which the coordinator then logs as an Error per daemon.
+        if (_disposed) return;
+
+        try
+        {
+            await _semaphore.WaitAsync(_cancellation.Token).ConfigureAwait(false);
+        }
+        catch (ObjectDisposedException)
+        {
+            // Dispose() raced this stop between the flag check and the token access; same benign
+            // "already disposed" outcome as the early return above.
+            return;
+        }
 
         try
         {

src/JasperFx.Events/Daemon/ProjectionCoordinatorBase.cs

@@ -95,6 +95,16 @@ protected ProjectionCoordinatorBase(
     /// </summary>
     protected abstract IReadOnlyList<IProjectionDaemon> ResolvedDaemons();
 
+    /// <summary>
+    /// Drop every daemon from the subclass cache. Called by <see cref="StopAsync"/> after it has
+    /// disposed the resolved daemons (marten#5055): without this, the cache keeps handing back
+    /// disposed daemons — a second Pause/Stop fans <c>StopAllAsync</c> out over them (one
+    /// ObjectDisposedException error log per daemon), and a later <see cref="ResumeAsync"/> or
+    /// daemon accessor returns a dead instance instead of rebuilding a fresh one. After this runs,
+    /// <see cref="ResolvedDaemons"/> must return empty.
+    /// </summary>
+    protected abstract void ClearResolvedDaemons();
+
     /// <inheritdoc />
     public abstract IProjectionDaemon DaemonForMainDatabase();
 
@@ -139,6 +149,14 @@ public async Task PauseAsync()
             {
                 await daemon.StopAllAsync().ConfigureAwait(false);
             }
+            catch (ObjectDisposedException exception)
+            {
+                // marten#5055: at shutdown a second Pause/Stop can fan out over daemons the first
+                // pass already disposed. There is nothing left to stop, so this is not an error —
+                // same philosophy as the jasperfx#499 disposed-data-source handling in executeAsync.
+                _logger.LogDebug(exception,
+                    "Projection daemon was already disposed while pausing; this is benign during shutdown");
+            }
             catch (Exception exception)
             {
                 _logger.LogError(exception, "Error while trying to stop daemon agents");
@@ -162,6 +180,10 @@ public virtual async Task StopAsync(CancellationToken cancellationToken)
             daemon.SafeDispose();
         }
 
+        // marten#5055: the daemons above are dead; purge them from the subclass cache so a second
+        // StopAsync has nothing to fan out over and a later ResumeAsync builds fresh daemons.
+        ClearResolvedDaemons();
+
         var distributor = Distributor;
         if (distributor != null)
         {