diff --git a/Directory.Packages.props b/Directory.Packages.props index 43ef228571..bec8b80b1b 100644 --- a/Directory.Packages.props +++ b/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. --> - - - + 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). --> + + + all runtime; build; native; contentfiles; analyzers; buildtransitive - + diff --git a/src/DaemonTests/Bug_5055_5056_coordinator_double_stop.cs b/src/DaemonTests/Bug_5055_5056_coordinator_double_stop.cs new file mode 100644 index 0000000000..2d280f2f8d --- /dev/null +++ b/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 +{ + 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)) + .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(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 only + // echoes to output. Thread-safe: the coordinator loop and the test body log concurrently. + private sealed class CapturingLogger: ILogger + { + 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 state) where TState : notnull => NullScope.Instance; + + public bool IsEnabled(LogLevel logLevel) => true; + + public void Log(LogLevel logLevel, EventId eventId, TState state, Exception? exception, + Func 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() { } + } + } +} diff --git a/src/Marten/Events/Daemon/Coordination/ExplicitProjectionCoordinator.cs b/src/Marten/Events/Daemon/Coordination/ExplicitProjectionCoordinator.cs index 8743965e20..3bd1993ec6 100644 --- a/src/Marten/Events/Daemon/Coordination/ExplicitProjectionCoordinator.cs +++ b/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.Empty; } } diff --git a/src/Marten/Events/Daemon/Coordination/ProjectionCoordinator.cs b/src/Marten/Events/Daemon/Coordination/ProjectionCoordinator.cs index 6144a7881a..a0025ed65b 100644 --- a/src/Marten/Events/Daemon/Coordination/ProjectionCoordinator.cs +++ b/src/Marten/Events/Daemon/Coordination/ProjectionCoordinator.cs @@ -148,6 +148,17 @@ protected override IReadOnlyList 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.Empty; + } + } + public override IProjectionDaemon DaemonForMainDatabase() { var database = (MartenDatabase)Store.Tenancy.Default.Database; diff --git a/src/Marten/MartenServiceCollectionExtensions.cs b/src/Marten/MartenServiceCollectionExtensions.cs index 7e3058668b..4102a5c6d0 100644 --- a/src/Marten/MartenServiceCollectionExtensions.cs +++ b/src/Marten/MartenServiceCollectionExtensions.cs @@ -516,7 +516,13 @@ public MartenStoreExpression 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, ProjectionCoordinator>(); Services.AddSingleton(s => s.GetRequiredService>()); @@ -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(); Services.AddSingleton(s => s.GetRequiredService());