fix: implement two-phase dispose for MessageBus subscription lifecycle

[niemyjski]

Summary

• *Add \IAsyncDisposable\ to \IMessageBus* interface for proper async cleanup
• Implement two-phase dispose in \MessageBusBase: \ShutdownAsync\ (graceful drain) then \CleanupAsync\ (resource teardown)
• Add disposal guard in \SendMessageToSubscribersAsync\ to prevent message processing during dispose
• *Simplify \CancellationToken.Register* callbacks to only remove from subscriber dictionary
• *Add \NullMessageBus.DisposeAsync* to satisfy interface contract
• Add 6 shared lifecycle tests to \MessageBusTestBase\ (dispose without exceptions, idempotent dispose, cancel token without teardown, subscribe/publish after dispose, dispose while publishing)
• Update messaging docs with disposal lifecycle, message durability table by provider, and custom provider guidance

Test plan

• All 1845 Foundatio core tests pass (0 failures)
• CI verifies no regressions
• Provider PRs pass CI with integration tests

Related Provider PRs

• RabbitMQ: fix: adapt RabbitMQ MessageBus to two-phase dispose lifecycle Foundatio.RabbitMQ#73
• AWS (SQS): fix: adapt SQS MessageBus to two-phase dispose lifecycle Foundatio.AWS#400
• Azure Service Bus: fix: adapt Azure Service Bus MessageBus to two-phase dispose lifecycle Foundatio.AzureServiceBus#77
• Kafka: fix: adapt Kafka MessageBus to two-phase dispose lifecycle Foundatio.Kafka#54
• Redis: fix: adapt Redis MessageBus to two-phase dispose lifecycle Foundatio.Redis#161

[Copilot]

Pull request overview

Implements an async-friendly, two-phase disposal lifecycle for message buses to better control shutdown/drain behavior and standardize cleanup across providers and tests.

Changes:

• Extend IMessageBus to include IAsyncDisposable, and implement DisposeAsync across core message bus types.
• Add two-phase disposal hooks (ShutdownAsync then CleanupAsync) in MessageBusBase, plus a disposal guard during subscriber dispatch.
• Add shared lifecycle tests in MessageBusTestBase and wire them into InMemoryMessageBusTests; update messaging docs with the new lifecycle and provider behavior.

Reviewed changes

Copilot reviewed 7 out of 7 changed files in this pull request and generated 7 comments.

Show a summary per file

File	Description
tests/Foundatio.Tests/Messaging/InMemoryMessageBusTests.cs	Exposes new lifecycle tests for the in-memory provider via [Fact] overrides.
src/Foundatio/Messaging/NullMessageBus.cs	Adds DisposeAsync to satisfy the updated interface contract.
src/Foundatio/Messaging/MessageBusBase.cs	Adds IAsyncDisposable, two-phase disposal hooks, and disposal-time subscriber dispatch guard.
src/Foundatio/Messaging/InMemoryMessageBus.cs	Overrides DisposeAsync to clear message counts before delegating to base disposal.
src/Foundatio/Messaging/IMessageBus.cs	Updates the message bus interface to require IAsyncDisposable.
src/Foundatio.TestHarness/Messaging/MessageBusTestBase.cs	Adds shared async-disposal lifecycle tests for all message bus implementations.
docs/guide/messaging.md	Documents async disposal, two-phase lifecycle, and shutdown durability expectations per provider.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[Copilot]

Pull request overview

Copilot reviewed 7 out of 7 changed files in this pull request and generated 1 comment.

Comments suppressed due to low confidence (1)

src/Foundatio/Messaging/MessageBusBase.cs:263

• In SubscribeImplAsync, the cancellation callback is registered before the subscriber is added to _subscribers. If the token is canceled concurrently (after ThrowIfCancellationRequested in SubscribeAsync but before/while registering), the callback can run immediately, fail to remove (not yet added), and then the subsequent TryAdd leaves a canceled subscriber in the dictionary until the next message dispatch. Consider adding the subscriber to _subscribers before registering, or re-checking cancellationToken.IsCancellationRequested after TryAdd and removing immediately if it’s already canceled.

    protected virtual Task SubscribeImplAsync<T>(Func<T, CancellationToken, Task> handler, CancellationToken cancellationToken) where T : class
    {
        var subscriber = new Subscriber
        {
            CancellationToken = cancellationToken,
            Type = typeof(T),
            Action = (message, token) =>
            {
                if (message is T typedMessage)
                    return handler(typedMessage, token);

                if (message is null)
                {
                    _logger.LogWarning("Subscriber action skipped: message body is null (likely a deserialization failure) for subscriber type {SubscriberType}", typeof(T));
                    return Task.CompletedTask;
                }

                _logger.LogTrace("Unable to call subscriber action: {MessageType} cannot be safely casted to {SubscriberType}", message.GetType(), typeof(T));
                return Task.CompletedTask;
            }
        };

        if (cancellationToken != CancellationToken.None)
        {
            // CancellationToken.Register only accepts synchronous callbacks, so we cannot safely
            // call RemoveTopicSubscriptionAsync here. The async-capable CancelAsync was added in
            // .NET 8 but still does not support async callbacks — see:
            // https://github.com/dotnet/runtime/issues/31315
            // Topic subscription teardown is handled during DisposeAsync via ShutdownAsync/CleanupAsync.
            cancellationToken.Register(() =>
            {
                _subscribers.TryRemove(subscriber.Id, out _);
            });
        }

        if (subscriber.Type.Name == "IMessage`1" && subscriber.Type.GenericTypeArguments.Length == 1)
        {
            var modelType = subscriber.Type.GenericTypeArguments.Single();
            subscriber.GenericType = typeof(Message<>).MakeGenericType(modelType);
        }

        if (!_subscribers.TryAdd(subscriber.Id, subscriber))
            _logger.LogError("Unable to add subscriber {SubscriberId}", subscriber.Id);

        return Task.CompletedTask;
    }

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[Copilot]

SubscribeAsync_CancelledToken_DoesNotTearDownInfrastructureAsync doesn’t actually assert that infrastructure teardown didn’t occur; it only verifies that subscribing/publishing still works afterwards (which would also pass if infrastructure was torn down and recreated). Also, the Task.Delay(100) introduces timing-based flakiness. Consider asserting directly that the canceled subscriber is not invoked after cancellation (e.g., publish after cancel and assert the first countdown is unchanged), and replace the fixed delay with a deterministic wait (signal from the handler, polling with timeout, etc.).

Suggested change

	var countdown = new AsyncCountdownEvent(1);
	await messageBus.SubscribeAsync<SimpleMessageA>(msg =>
	{
	Assert.Equal("Hello", msg.Data);
	countdown.Signal();
	}, cts.Token);
	await cts.CancelAsync();
	await Task.Delay(100, TestCancellationToken);
	var countdown2 = new AsyncCountdownEvent(1);
	await messageBus.SubscribeAsync<SimpleMessageA>(msg =>
	{
	Assert.Equal("Hello", msg.Data);
	countdown2.Signal();
	}, TestCancellationToken);
	await messageBus.PublishAsync(new SimpleMessageA { Data = "Hello" }, cancellationToken: TestCancellationToken);
	await countdown2.WaitAsync(TimeSpan.FromSeconds(5));
	Assert.Equal(0, countdown2.CurrentCount);
	long cancelledSubscriberInvocations = 0;
	await messageBus.SubscribeAsync<SimpleMessageA>(msg =>
	{
	Assert.Equal("Hello", msg.Data);
	Interlocked.Increment(ref cancelledSubscriberInvocations);
	}, cts.Token);
	await cts.CancelAsync();
	var activeSubscriberCountdown = new AsyncCountdownEvent(1);
	await messageBus.SubscribeAsync<SimpleMessageA>(msg =>
	{
	Assert.Equal("Hello", msg.Data);
	activeSubscriberCountdown.Signal();
	}, TestCancellationToken);
	await messageBus.PublishAsync(new SimpleMessageA { Data = "Hello" }, cancellationToken: TestCancellationToken);
	await activeSubscriberCountdown.WaitAsync(TimeSpan.FromSeconds(5));
	Assert.Equal(0, activeSubscriberCountdown.CurrentCount);
	Assert.Equal(0, Interlocked.Read(ref cancelledSubscriberInvocations));

[niemyjski]

Acknowledged. The test verifies that a cancelled subscription token does not tear down the bus infrastructure by confirming that re-subscribing and publishing still work. This is a reasonable integration-level assertion for the behavior being tested. The Task.Delay is used to give the cancellation time to propagate — we can replace with a more deterministic approach in a follow-up if flakiness is observed.