Use DisposedCancellationToken for message bus topic/subscription setup

EnsureTopicCreatedAsync and EnsureTopicSubscriptionAsync are one-time infrastructure setup methods. Previously they received the caller's cancellation token (via the linked token in PublishAsync, or directly in SubscribeAsync). This meant that if an individual caller cancelled their publish or subscribe operation, topic creation or subscription setup could be aborted mid-way, leaving infrastructure in a half-created state for subsequent callers — the same class of bug fixed in queue creation (99e3353).

Now both setup methods receive DisposedCancellationToken, ensuring they only abort when the message bus itself is being disposed. The actual publish and subscribe operations continue to use the linked/caller token so they remain individually cancellable.

Affected concrete implementations (Azure Service Bus, RabbitMQ, SQS, Kafka, Redis) all receive the token through the overridden method parameter, so their internal lock acquisitions and network calls automatically benefit from this fix.

Co-authored-by: Cursor <cursoragent@cursor.com>

Author: niemyjski

src/Foundatio/Messaging/MessageBusBase.cs

@@ -87,7 +87,9 @@ public async Task PublishAsync(Type messageType, object message, MessageOptions
         using var linkedCancellationTokenSource = GetLinkedDisposableCancellationTokenSource(cancellationToken);
         try
         {
-            await EnsureTopicCreatedAsync(linkedCancellationTokenSource.Token).AnyContext();
+            // Use DisposedCancellationToken for setup: topic creation should only abort on disposal,
+            // not due to an individual caller's cancellation token.
+            await EnsureTopicCreatedAsync(DisposedCancellationToken).AnyContext();
             await PublishImplAsync(GetMappedMessageType(messageType), message, options, linkedCancellationTokenSource.Token).AnyContext();
         }
         catch (Exception ex) when (ex is not OperationCanceledException and not MessageBusException)
@@ -194,7 +196,9 @@ public async Task SubscribeAsync<T>(Func<T, CancellationToken, Task> handler, Ca
         _logger.LogTrace("Adding subscriber for {MessageType}", typeof(T).FullName);
 
         await SubscribeImplAsync(handler, cancellationToken).AnyContext();
-        await EnsureTopicSubscriptionAsync(cancellationToken).AnyContext();
+        // Use DisposedCancellationToken for setup: subscription infrastructure should only abort on disposal,
+        // not due to the caller's cancellation token.
+        await EnsureTopicSubscriptionAsync(DisposedCancellationToken).AnyContext();
     }
 
     protected List<Subscriber> GetMessageSubscribers(IMessage message)