diff --git a/src/EventTests/Daemon/DisposedDaemonStopAllTests.cs b/src/EventTests/Daemon/DisposedDaemonStopAllTests.cs new file mode 100644 index 0000000..3755876 --- /dev/null +++ b/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> BuildDaemon() + { + var store = Substitute.For>(); + store.Meter.Returns(new Meter("tests")); + store.TimeProvider.Returns(TimeProvider.System); + + var database = Substitute.For(); + database.Identifier.Returns("db1"); + database.DatabaseUri.Returns(new Uri("fake://db1")); + database.Tracker.Returns(new ShardStateTracker(new NulloLogger())); + + return new JasperFxAsyncDaemon>( + 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 Detect(CancellationToken token) + => Task.FromResult(new HighWaterStatistics()); + + public Task DetectInSafeZone(CancellationToken token) => Detect(token); + } + + // Minimal concrete ProjectionGraph — on these paths the daemon consumes it only as DaemonSettings. + private sealed class FakeProjectionGraph : + ProjectionGraph, FakeOperations, FakeSession> + { + public FakeProjectionGraph() : base(Substitute.For(), "tests") + { + } + + protected override void onAddProjection(object projection) + { + // Nothing + } + } +} diff --git a/src/EventTests/Daemon/ProjectionCoordinatorBaseTests.cs b/src/EventTests/Daemon/ProjectionCoordinatorBaseTests.cs index 9c12c7f..1581b89 100644 --- a/src/EventTests/Daemon/ProjectionCoordinatorBaseTests.cs +++ b/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 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 Errors { get; } = []; + public List Debugs { get; } = []; public void Log(LogLevel logLevel, EventId eventId, TState state, Exception? exception, Func formatter) @@ -350,6 +420,13 @@ public void Log(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 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 VisibleResolvedDaemons => ResolvedDaemons(); + public override IProjectionDaemon DaemonForMainDatabase() => _daemon; public override ValueTask 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 _agents = []; private bool _failed; @@ -579,6 +665,11 @@ public IReadOnlyList CurrentAgents() public Task StopAllAsync() { + if (ThrowObjectDisposedOnStopAllWhenDisposed && DisposeCount > 0) + { + throw new ObjectDisposedException(nameof(CancellationTokenSource)); + } + StopAllCount++; return Task.CompletedTask; } diff --git a/src/JasperFx.Events/Daemon/JasperFxAsyncDaemon.cs b/src/JasperFx.Events/Daemon/JasperFxAsyncDaemon.cs index 1a41b4b..b076891 100644 --- a/src/JasperFx.Events/Daemon/JasperFxAsyncDaemon.cs +++ b/src/JasperFx.Events/Daemon/JasperFxAsyncDaemon.cs @@ -41,6 +41,11 @@ public partial class JasperFxAsyncDaemon _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 { diff --git a/src/JasperFx.Events/Daemon/ProjectionCoordinatorBase.cs b/src/JasperFx.Events/Daemon/ProjectionCoordinatorBase.cs index 78e0b19..f354985 100644 --- a/src/JasperFx.Events/Daemon/ProjectionCoordinatorBase.cs +++ b/src/JasperFx.Events/Daemon/ProjectionCoordinatorBase.cs @@ -95,6 +95,16 @@ protected ProjectionCoordinatorBase( /// protected abstract IReadOnlyList ResolvedDaemons(); + /// + /// Drop every daemon from the subclass cache. Called by after it has + /// disposed the resolved daemons (marten#5055): without this, the cache keeps handing back + /// disposed daemons — a second Pause/Stop fans StopAllAsync out over them (one + /// ObjectDisposedException error log per daemon), and a later or + /// daemon accessor returns a dead instance instead of rebuilding a fresh one. After this runs, + /// must return empty. + /// + protected abstract void ClearResolvedDaemons(); + /// 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) {