diff --git a/src/Core/src/Eventuous.Subscriptions/Channels/ChannelWorkerBase.cs b/src/Core/src/Eventuous.Subscriptions/Channels/ChannelWorkerBase.cs index 6f0f60109..3439e6e25 100644 --- a/src/Core/src/Eventuous.Subscriptions/Channels/ChannelWorkerBase.cs +++ b/src/Core/src/Eventuous.Subscriptions/Channels/ChannelWorkerBase.cs @@ -6,9 +6,12 @@ namespace Eventuous.Subscriptions.Channels; abstract class ChannelWorkerBase : IAsyncDisposable { - readonly CancellationTokenSource _cts = new(); + readonly CancellationTokenSource _cts = new(); + readonly TaskCompletionSource _disposed = new(TaskCreationOptions.RunContinuationsAsynchronously); readonly Task[] _readerTasks; + int _disposing; + public Func? OnDispose { get; set; } public ValueTask Write(T element, CancellationToken cancellationToken) @@ -24,16 +27,33 @@ protected ChannelWorkerBase(Channel channel, Func pr _readerTasks = Enumerable.Range(0, concurrencyLevel).Select(_ => Task.Run(() => processor(_cts.Token))).ToArray(); } - public async ValueTask DisposeAsync() { - _stopping = true; - await _channel.Stop(_cts, _readerTasks, OnDispose).NoContext(); + /// + /// Idempotent. The commit handler worker is disposed by both the resubscribe and the shutdown + /// paths, which can run concurrently, so a second call is expected rather than a programming + /// error. It must not re-enter the shutdown: by then the CTS is disposed, and cancelling it + /// again throws out of host shutdown. The second caller + /// awaits the first call's shutdown, so it can't return before the final checkpoint flush. + /// + public ValueTask DisposeAsync() => Interlocked.Exchange(ref _disposing, 1) == 0 ? new(StopWorker()) : new(_disposed.Task); + + async Task StopWorker() { + try { + _stopping = true; + await _channel.Stop(_cts, _readerTasks, OnDispose).NoContext(); #if NET8_0_OR_GREATER - await _cts.CancelAsync().NoContext(); + await _cts.CancelAsync().NoContext(); #else - _cts.Cancel(); + _cts.Cancel(); #endif - await Task.WhenAll(_readerTasks).NoThrow(); - _cts.Dispose(); - GC.SuppressFinalize(this); + await Task.WhenAll(_readerTasks).NoThrow(); + _cts.Dispose(); + GC.SuppressFinalize(this); + _disposed.TrySetResult(); + } catch (Exception e) { + // Don't let a waiter see a clean shutdown that didn't happen. + _disposed.TrySetException(e); + + throw; + } } } diff --git a/src/Core/src/Eventuous.Subscriptions/EventSubscription.cs b/src/Core/src/Eventuous.Subscriptions/EventSubscription.cs index 281dfc7e3..a3b67685d 100644 --- a/src/Core/src/Eventuous.Subscriptions/EventSubscription.cs +++ b/src/Core/src/Eventuous.Subscriptions/EventSubscription.cs @@ -212,12 +212,27 @@ protected void Dropped(DropReason reason, Exception? exception) { IsDropped = true; _onDropped?.Invoke(Options.SubscriptionId, reason, exception); + // Read the token here rather than inside the background task below: Unsubscribe disposes + // Stopping, and reading .Token from a disposed source throws, which the task would surface as + // a spurious warning plus an unobserved exception. A token captured before the dispose stays + // usable afterwards, so hoisting the read is what makes the resubscribe safe. Losing the race + // outright means shutdown already got there, and there's nothing left to resubscribe to. + CancellationToken stopping; + + try { stopping = Stopping.Token; } catch (ObjectDisposedException) { return; } + + // Same reasoning for a token that's merely cancelled, which is the state Unsubscribe leaves it + // in for most of shutdown. Resubscribing from there can't succeed, and it isn't free: the + // checkpoint subscription's Resubscribe disposes the commit handler before it ever looks at the + // token, putting a second disposer in the race with Finalize. + if (stopping.IsCancellationRequested) return; + Task.Run( async () => { var delay = reason == DropReason.Stopped ? TimeSpan.FromSeconds(10) : TimeSpan.FromSeconds(2); Log.SubscriptionWillResubscribe(delay); - try { await Resubscribe(delay, Stopping.Token).NoContext(); } catch (Exception e) { + try { await Resubscribe(delay, stopping).NoContext(); } catch (Exception e) { Log.WarnLog?.Log(e.Message); throw; diff --git a/src/Core/src/Eventuous.Subscriptions/EventSubscriptionWithCheckpoint.cs b/src/Core/src/Eventuous.Subscriptions/EventSubscriptionWithCheckpoint.cs index 863991299..c2871f623 100644 --- a/src/Core/src/Eventuous.Subscriptions/EventSubscriptionWithCheckpoint.cs +++ b/src/Core/src/Eventuous.Subscriptions/EventSubscriptionWithCheckpoint.cs @@ -142,7 +142,10 @@ protected override async Task Resubscribe(TimeSpan delay, CancellationToken canc protected override async ValueTask Finalize(CancellationToken cancellationToken) => await DisposeCommitHandler(); async ValueTask DisposeCommitHandler() { - // Swap to null first so the concurrent path (Resubscribe vs Finalize) sees null. + // Swap to null first so the concurrent path (Resubscribe vs Finalize) sees null. The read and + // the write aren't atomic, so both paths can still come away with the same handler — that stays + // safe because the commit worker's dispose is idempotent, and the second caller awaits the first + // one's shutdown rather than re-entering it and cancelling an already-disposed CTS (AI-1699). var handler = CheckpointCommitHandler; CheckpointCommitHandler = null; diff --git a/src/Core/test/Eventuous.Tests.Subscriptions/CheckpointCommitHandlerDisposeTests.cs b/src/Core/test/Eventuous.Tests.Subscriptions/CheckpointCommitHandlerDisposeTests.cs new file mode 100644 index 000000000..2ee96c988 --- /dev/null +++ b/src/Core/test/Eventuous.Tests.Subscriptions/CheckpointCommitHandlerDisposeTests.cs @@ -0,0 +1,58 @@ +using Eventuous.Subscriptions.Checkpoints; +using Shouldly; + +namespace Eventuous.Tests.Subscriptions; + +/// +/// Guards against AI-1699, where host shutdown failed with +/// ObjectDisposedException: The CancellationTokenSource has been disposed thrown from +/// ChannelWorkerBase.DisposeAsync via . +/// The commit handler can be disposed twice — Resubscribe and Finalize both call +/// DisposeCommitHandler, and the second call must be a no-op rather than a crash. +/// +public class CheckpointCommitHandlerDisposeTests { + [Test] + public async Task Dispose_is_idempotent() { + var store = new NoOpCheckpointStore(); + var handler = new CheckpointCommitHandler("test-dispose-twice", store, TimeSpan.FromMilliseconds(10)); + + await handler.DisposeAsync(); + + await Should.NotThrowAsync(async () => await handler.DisposeAsync()); + } + + /// + /// The loser of the dispose race must not report the worker stopped before it actually is — + /// host shutdown continues on that return, and the final checkpoint flush is still in flight. + /// + [Test] + public async Task Second_dispose_awaits_the_first(CancellationToken ct) { + var storeEntered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var release = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + + var handler = new CheckpointCommitHandler( + "test-dispose-awaits", + async (checkpoint, _, _) => { + storeEntered.TrySetResult(); + await release.Task; + + return checkpoint; + }, + TimeSpan.FromMilliseconds(1) + ); + + // Park the commit worker inside the checkpoint store so the first dispose can't complete. + await handler.Commit(new CommitPosition(0, 0, DateTime.UtcNow), ct); + await storeEntered.Task.WaitAsync(TimeSpan.FromSeconds(10), ct); + + var first = handler.DisposeAsync(); + var second = handler.DisposeAsync(); + + await Task.Delay(200, ct); + second.IsCompleted.ShouldBeFalse("the second dispose must await the shutdown in flight, not skip past it"); + + release.SetResult(); + await first; + await second; + } +} diff --git a/src/Core/test/Eventuous.Tests.Subscriptions/SubscriptionShutdownTests.cs b/src/Core/test/Eventuous.Tests.Subscriptions/SubscriptionShutdownTests.cs new file mode 100644 index 000000000..cacf848a0 --- /dev/null +++ b/src/Core/test/Eventuous.Tests.Subscriptions/SubscriptionShutdownTests.cs @@ -0,0 +1,123 @@ +using System.Collections.Concurrent; +using Eventuous.Subscriptions; +using Eventuous.Subscriptions.Filters; +using Microsoft.Extensions.Logging; +using Shouldly; + +namespace Eventuous.Tests.Subscriptions; + +/// +/// A dropped subscription schedules the resubscribe on a background task, and that task used to read +/// Stopping.Token long after Unsubscribe got to the source. A drop that races shutdown is +/// benign — there is nothing left to resubscribe to — but it used to cost a spurious warning, an +/// unobserved exception, and a commit-handler dispose racing the one in Finalize (AI-1699). +/// +public class SubscriptionShutdownTests { + /// + /// The KurrentDB subscriptions cancel Stopping at the top of their Unsubscribe, so a + /// drop during shutdown normally finds the token cancelled. Resubscribing from there is pure waste: + /// EventSubscriptionWithCheckpoint.Resubscribe disposes the commit handler before it ever + /// looks at the token, which is what put a second disposer in the race with Finalize. + /// + [Test] + public async Task Drop_after_shutdown_started_does_not_resubscribe(CancellationToken ct) { + var subscription = new TestSubscription(new() { SubscriptionId = "test-drop-when-cancelled" }, new ConsumePipe(), new CapturingLoggerFactory()); + + await subscription.Subscribe(_ => { }, (_, _, _) => { }, ct); + + subscription.DropAfterCancellingStopping(); + + var resubscribed = await subscription.WaitForResubscribe(TimeSpan.FromSeconds(2)); + + resubscribed.ShouldBeFalse("a subscription that is already stopping has nothing to resubscribe to"); + } + + [Test] + public async Task Drop_racing_unsubscribe_does_not_report_a_disposed_cts(CancellationToken ct) { + var logs = new CapturingLoggerFactory(); + + var subscription = new TestSubscription(new() { SubscriptionId = "test-drop-race" }, new ConsumePipe(), logs); + + await subscription.Subscribe(_ => { }, (_, _, _) => { }, ct); + + // Reproduce the losing side of the race: Unsubscribe has already disposed Stopping while the + // subscription still believes it's running, which is exactly what lets Dropped reach the token. + subscription.DropAfterDisposingStopping(); + + var reported = await logs.WaitForWarning("CancellationTokenSource has been disposed", TimeSpan.FromSeconds(2)); + + reported.ShouldBeFalse("dropping while Unsubscribe disposes Stopping is a benign shutdown race, not an error"); + } + + record TestSubscriptionOptions : SubscriptionOptions; + + /// + /// A subscription that does nothing but expose the drop path. Stopping is protected, so both + /// shutdown states can be reproduced without any test-only hooks in the production class. + /// + class TestSubscription(TestSubscriptionOptions options, ConsumePipe pipe, ILoggerFactory loggerFactory) + : EventSubscription(options, pipe, loggerFactory, null) { + readonly TaskCompletionSource _resubscribed = new(TaskCreationOptions.RunContinuationsAsynchronously); + + public void DropAfterCancellingStopping() { + Stopping.Cancel(false); + Drop(); + } + + public void DropAfterDisposingStopping() { + Stopping.Dispose(); + Drop(); + } + + public async Task WaitForResubscribe(TimeSpan timeout) + => await Task.WhenAny(_resubscribed.Task, Task.Delay(timeout)) == _resubscribed.Task; + + protected override Task Resubscribe(TimeSpan delay, CancellationToken cancellationToken) { + _resubscribed.TrySetResult(); + + return Task.CompletedTask; + } + + protected override ValueTask Subscribe(CancellationToken cancellationToken) => default; + + protected override ValueTask Unsubscribe(CancellationToken cancellationToken) => default; + + void Drop() => Dropped(DropReason.SubscriptionError, new InvalidOperationException("Simulated drop during shutdown")); + } + + sealed class CapturingLoggerFactory : ILoggerFactory { + readonly ConcurrentQueue _warnings = []; + + /// + /// Polls rather than waiting out the full timeout, so the failing case reports in milliseconds. + /// The resubscribe runs on a fire-and-forget task, so there is nothing to await on. + /// + public async Task WaitForWarning(string contains, TimeSpan timeout) { + var deadline = DateTime.UtcNow + timeout; + + while (DateTime.UtcNow < deadline) { + if (_warnings.Any(w => w.Contains(contains))) return true; + + await Task.Delay(20); + } + + return false; + } + + public ILogger CreateLogger(string categoryName) => new CapturingLogger(_warnings); + + public void AddProvider(ILoggerProvider provider) { } + + public void Dispose() { } + + sealed class CapturingLogger(ConcurrentQueue warnings) : ILogger { + public IDisposable? BeginScope(TState state) where TState : notnull => null; + + public bool IsEnabled(LogLevel logLevel) => true; + + public void Log(LogLevel logLevel, EventId eventId, TState state, Exception? exception, Func formatter) { + if (logLevel >= LogLevel.Warning) warnings.Enqueue(formatter(state, exception)); + } + } + } +}