Add transport-native temporary receive endpoints to Mocha

src/Mocha/src/Mocha.Transport.AzureServiceBus/AzureServiceBusReceiveEndpoint.cs

@@ -97,8 +97,9 @@ protected override void OnComplete(
 
     /// <summary>
     /// Starts when the messaging runtime activates this endpoint. The endpoint selects and starts
-    /// the appropriate session or non-session processor, and starts a heartbeat for reply queues.
-    /// If startup fails, it disposes any resources created during the attempt before propagating the failure.
+    /// the appropriate session or non-session processor, and starts a heartbeat for queues that
+    /// have an idle deletion window configured. If startup fails, it disposes any resources created
+    /// during the attempt before propagating the failure.
     /// </summary>
     protected override async ValueTask OnStartAsync(
         IMessagingRuntimeContext context,
@@ -119,10 +120,10 @@ protected override async ValueTask OnStartAsync(
 
             await _processor.StartProcessingAsync(cancellationToken);
 
-            if (Configuration.Kind == ReceiveEndpointKind.Reply)
+            if (Queue.AutoDeleteOnIdle is { } autoDeleteOnIdle)
             {
                 var receiver = transport.ClientManager.CreateReceiver(Queue.Name);
-                _heartbeat = new QueueHeartbeat(receiver, _logger, Queue.Name);
+                _heartbeat = new QueueHeartbeat(receiver, autoDeleteOnIdle, _logger, Queue.Name);
             }
         }
         catch
@@ -218,7 +219,8 @@ private Task OnProcessorError(ProcessErrorEventArgs args)
 
     /// <summary>
     /// Stops when the messaging runtime deactivates this endpoint, then disposes the processor and
-    /// heartbeat resources.
+    /// heartbeat resources. For a temporary endpoint, also removes the Mocha-owned, auto-provisioned
+    /// forwarding subscriptions that route to this endpoint's queue.
     /// </summary>
     protected override async ValueTask OnStopAsync(
         IMessagingRuntimeContext context,
@@ -230,6 +232,11 @@ protected override async ValueTask OnStopAsync(
             {
                 await _processor.StopProcessingAsync(cancellationToken);
             }
+
+            if (Configuration.IsTemporary)
+            {
+                await CleanupTemporaryResourcesAsync(cancellationToken);
+            }
         }
         finally
         {
@@ -247,6 +254,54 @@ protected override async ValueTask OnStopAsync(
         }
     }
 
+    /// <summary>
+    /// Removes convention forwarding subscriptions for this temporary endpoint. The queue is
+    /// removed when no provisioned declared subscriptions remain and all convention cleanup succeeds.
+    /// </summary>
+    private async Task CleanupTemporaryResourcesAsync(CancellationToken cancellationToken)
+    {
+        var topology = (AzureServiceBusMessagingTopology)transport.Topology;
+        var canDeleteQueue = true;
+
+        foreach (var subscription in Queue.Subscriptions)
+        {
+            if (!(subscription.AutoProvision ?? topology.AutoProvision))
+            {
+                continue;
+            }
+
+            if (subscription.Origin != TopologyOrigin.Convention)
+            {
+                canDeleteQueue = false;
+                continue;
+            }
+
+            try
+            {
+                await subscription.DeprovisionAsync(transport.ClientManager, cancellationToken);
+            }
+            catch (Exception ex) when (ex is not OperationCanceledException)
+            {
+                canDeleteQueue = false;
+                _logger.ForwardingSubscriptionCleanupFailed(ex, subscription.Source.Name, subscription.Name);
+            }
+        }
+
+        if (!canDeleteQueue || !(Queue.AutoProvision ?? topology.AutoProvision))
+        {
+            return;
+        }
+
+        try
+        {
+            await Queue.DeprovisionAsync(transport.ClientManager, cancellationToken);
+        }
+        catch (Exception ex) when (ex is not OperationCanceledException)
+        {
+            _logger.TemporaryQueueCleanupFailed(ex, Queue.Name);
+        }
+    }
+
     private static bool IsTransientProcessorError(Exception exception)
     {
         if (exception is OperationCanceledException)
@@ -307,4 +362,19 @@ public static partial void ProcessorError(
 
     [LoggerMessage(LogLevel.Warning, "Reply queue keep-alive peek failed for {EntityPath}")]
     public static partial void ReplyQueueKeepAliveFailed(this ILogger logger, Exception exception, string entityPath);
+
+    [LoggerMessage(
+        LogLevel.Warning,
+        "Failed to delete Azure Service Bus forwarding subscription '{SubscriptionName}' on topic '{TopicName}'")]
+    public static partial void ForwardingSubscriptionCleanupFailed(
+        this ILogger logger,
+        Exception exception,
+        string topicName,
+        string subscriptionName);
+
+    [LoggerMessage(LogLevel.Warning, "Failed to delete temporary Azure Service Bus queue '{QueueName}'")]
+    public static partial void TemporaryQueueCleanupFailed(
+        this ILogger logger,
+        Exception exception,
+        string queueName);
 }

src/Mocha/src/Mocha.Transport.AzureServiceBus/Configurations/AzureServiceBusQueueDescriptorConfiguration.cs

@@ -45,6 +45,16 @@ internal AzureServiceBusQueueDescriptorConfiguration(string name)
     /// </summary>
     public ReceiveEndpointKind? Kind { get; set; }
 
+    /// <summary>
+    /// Gets or sets whether the queue's receive endpoint is temporary.
+    /// </summary>
+    public bool IsTemporary { get; set; }
+
+    /// <summary>
+    /// Gets or sets the idle window after which a temporary queue may be deleted by the broker.
+    /// </summary>
+    public TimeSpan? TemporaryIdleTimeout { get; set; }
+
     /// <summary>
     /// Gets or sets the receive endpoint maximum concurrency.
     /// </summary>

src/Mocha/src/Mocha.Transport.AzureServiceBus/Configurations/AzureServiceBusReceiveEndpointConfiguration.cs

@@ -50,4 +50,25 @@ public sealed class AzureServiceBusReceiveEndpointConfiguration : ReceiveEndpoin
     /// the endpoint default of five minutes is used.
     /// </summary>
     public TimeSpan? MaxAutoLockRenewalDuration { get; set; }
+
+    /// <summary>
+    /// Gets or sets the idle window after which a temporary endpoint's queue may be deleted by
+    /// the broker. When <see langword="null"/> and <see cref="ReceiveEndpointConfiguration.IsTemporary"/>
+    /// is set, <see cref="TemporaryDefaults.AutoDeleteOnIdle"/> applies.
+    /// </summary>
+    public TimeSpan? TemporaryIdleTimeout { get; set; }
+
+    public static class TemporaryDefaults
+    {
+        /// <summary>
+        /// The default idle window applied to a temporary endpoint marked via the parameterless
+        /// <c>Temporary()</c> overload.
+        /// </summary>
+        public static readonly TimeSpan AutoDeleteOnIdle = TimeSpan.FromHours(24);
+
+        /// <summary>
+        /// The minimum idle window Azure Service Bus accepts for <c>AutoDeleteOnIdle</c>.
+        /// </summary>
+        public static readonly TimeSpan MinimumAutoDeleteOnIdle = TimeSpan.FromMinutes(5);
+    }
 }

src/Mocha/src/Mocha.Transport.AzureServiceBus/Connection/AzureServiceBusClientManager.cs

@@ -210,6 +210,24 @@ public Task CreateSubscriptionAsync(CreateSubscriptionOptions options, Cancellat
         return _connection.CreateSubscriptionAsync(options, cancellationToken);
     }
 
+    /// <summary>
+    /// Deletes a queue using the administration client.
+    /// </summary>
+    public Task DeleteQueueAsync(string queueName, CancellationToken cancellationToken)
+    {
+        ObjectDisposedException.ThrowIf(_isDisposed, this);
+        return _connection.DeleteQueueAsync(queueName, cancellationToken);
+    }
+
+    /// <summary>
+    /// Deletes a subscription using the administration client.
+    /// </summary>
+    public Task DeleteSubscriptionAsync(string topicName, string subscriptionName, CancellationToken cancellationToken)
+    {
+        ObjectDisposedException.ThrowIf(_isDisposed, this);
+        return _connection.DeleteSubscriptionAsync(topicName, subscriptionName, cancellationToken);
+    }
+
     /// <inheritdoc />
     public async ValueTask DisposeAsync()
     {

src/Mocha/src/Mocha.Transport.AzureServiceBus/Connection/ServiceBusConnection.cs

@@ -56,6 +56,12 @@ public Task CreateTopicAsync(CreateTopicOptions options, CancellationToken cance
     public Task CreateSubscriptionAsync(CreateSubscriptionOptions options, CancellationToken cancellationToken)
         => AdministrationClient.CreateSubscriptionAsync(options, cancellationToken);
 
+    public Task DeleteQueueAsync(string queueName, CancellationToken cancellationToken)
+        => AdministrationClient.DeleteQueueAsync(queueName, cancellationToken);
+
+    public Task DeleteSubscriptionAsync(string topicName, string subscriptionName, CancellationToken cancellationToken)
+        => AdministrationClient.DeleteSubscriptionAsync(topicName, subscriptionName, cancellationToken);
+
     public ValueTask DisposeAsync() => _ownsClient ? _client.DisposeAsync() : ValueTask.CompletedTask;
 
     private ServiceBusAdministrationClient AdministrationClient

src/Mocha/src/Mocha.Transport.AzureServiceBus/Descriptors/AzureServiceBusMessagingTransportDescriptor.cs

@@ -363,6 +363,8 @@ private void ConfigureQueueEndpoint(AzureServiceBusQueueDescriptorConfiguration
         }
 
         target.UseNativeDeadLetterForwarding |= configuration.UseNativeDeadLetterForwarding;
+        target.IsTemporary |= configuration.IsTemporary;
+        target.TemporaryIdleTimeout ??= configuration.TemporaryIdleTimeout;
         target.MaxConcurrentSessions ??= configuration.MaxConcurrentSessions;
         target.MaxConcurrentCallsPerSession ??= configuration.MaxConcurrentCallsPerSession;
         target.SessionIdleTimeout ??= configuration.SessionIdleTimeout;

src/Mocha/src/Mocha.Transport.AzureServiceBus/Descriptors/AzureServiceBusQueueDescriptor.cs

@@ -138,6 +138,23 @@ public IAzureServiceBusQueueDescriptor Kind(ReceiveEndpointKind kind)
         return this;
     }
 
+    public IAzureServiceBusQueueDescriptor Temporary()
+    {
+        Configuration.IsTemporary = true;
+        return this;
+    }
+
+    public IAzureServiceBusQueueDescriptor Temporary(TimeSpan idleTimeout)
+    {
+        ArgumentOutOfRangeException.ThrowIfLessThan(
+            idleTimeout,
+            AzureServiceBusReceiveEndpointConfiguration.TemporaryDefaults.MinimumAutoDeleteOnIdle);
+
+        Configuration.IsTemporary = true;
+        Configuration.TemporaryIdleTimeout = idleTimeout;
+        return this;
+    }
+
     public IAzureServiceBusQueueDescriptor MaxConcurrency(int maxConcurrency)
     {
         Configuration.MaxConcurrency = maxConcurrency;

src/Mocha/src/Mocha.Transport.AzureServiceBus/Descriptors/AzureServiceBusReceiveEndpointDescriptor.cs

@@ -92,6 +92,27 @@ internal AzureServiceBusReceiveEndpointDescriptor(IMessagingConfigurationContext
         return this;
     }
 
+    /// <inheritdoc />
+    public new IAzureServiceBusReceiveEndpointDescriptor Temporary()
+    {
+        base.Temporary();
+
+        return this;
+    }
+
+    /// <inheritdoc />
+    public IAzureServiceBusReceiveEndpointDescriptor Temporary(TimeSpan idleTimeout)
+    {
+        ArgumentOutOfRangeException.ThrowIfLessThan(
+            idleTimeout,
+            AzureServiceBusReceiveEndpointConfiguration.TemporaryDefaults.MinimumAutoDeleteOnIdle);
+
+        base.Temporary();
+        Configuration.TemporaryIdleTimeout = idleTimeout;
+
+        return this;
+    }
+
     /// <inheritdoc />
     public IAzureServiceBusReceiveEndpointDescriptor Queue(string name)
     {

src/Mocha/src/Mocha.Transport.AzureServiceBus/Descriptors/IAzureServiceBusQueueDescriptor.cs

@@ -49,6 +49,22 @@ IAzureServiceBusQueueDescriptor DeadLetteringOnMessageExpiration(
 
     IAzureServiceBusQueueDescriptor Kind(ReceiveEndpointKind kind);
 
+    /// <summary>
+    /// Marks this queue's receive endpoint as temporary, using the default idle window.
+    /// </summary>
+    /// <returns>The descriptor for method chaining.</returns>
+    IAzureServiceBusQueueDescriptor Temporary();
+
+    /// <summary>
+    /// Marks this queue's receive endpoint as temporary with an explicit idle window after which
+    /// the broker may delete the queue.
+    /// </summary>
+    /// <param name="idleTimeout">
+    /// The idle window. Must be at least <see cref="AzureServiceBusReceiveEndpointConfiguration.TemporaryDefaults.MinimumAutoDeleteOnIdle"/>.
+    /// </param>
+    /// <returns>The descriptor for method chaining.</returns>
+    IAzureServiceBusQueueDescriptor Temporary(TimeSpan idleTimeout);
+
     IAzureServiceBusQueueDescriptor MaxConcurrency(int maxConcurrency);
 
     IAzureServiceBusQueueDescriptor PrefetchCount(int? count);

src/Mocha/src/Mocha.Transport.AzureServiceBus/Descriptors/IAzureServiceBusReceiveEndpointDescriptor.cs

@@ -36,6 +36,19 @@ public interface IAzureServiceBusReceiveEndpointDescriptor
     /// <inheritdoc cref="IReceiveEndpointDescriptor{T}.Kind(ReceiveEndpointKind)"/>
     new IAzureServiceBusReceiveEndpointDescriptor Kind(ReceiveEndpointKind kind);
 
+    /// <inheritdoc cref="IReceiveEndpointDescriptor{T}.Temporary"/>
+    new IAzureServiceBusReceiveEndpointDescriptor Temporary();
+
+    /// <summary>
+    /// Marks this receive endpoint as temporary with an explicit idle window after which the
+    /// broker may delete its backing queue.
+    /// </summary>
+    /// <param name="idleTimeout">
+    /// The idle window. Must be at least <see cref="AzureServiceBusReceiveEndpointConfiguration.TemporaryDefaults.MinimumAutoDeleteOnIdle"/>.
+    /// </param>
+    /// <returns>The descriptor instance for method chaining.</returns>
+    IAzureServiceBusReceiveEndpointDescriptor Temporary(TimeSpan idleTimeout);
+
     /// <summary>
     /// Sets the absolute address of the fault endpoint.
     /// </summary>

src/Mocha/src/Mocha.Transport.AzureServiceBus/QueueHeartbeat.cs

@@ -4,12 +4,11 @@
 namespace Mocha.Transport.AzureServiceBus;
 
 /// <summary>
-/// Periodically peeks at a queue to reset its <c>AutoDeleteOnIdle</c> timer, keeping
-/// framework-managed reply queues alive even when no replies have flowed for the idle window.
+/// Periodically peeks at a queue to reset its <c>AutoDeleteOnIdle</c> timer, keeping a queue
+/// with an idle deletion window alive while its receive endpoint is active.
 /// </summary>
 internal sealed class QueueHeartbeat : IAsyncDisposable
 {
-    private static readonly TimeSpan s_defaultInterval = TimeSpan.FromHours(12);
     private static readonly TimeSpan s_stopTimeout = TimeSpan.FromSeconds(5);
 
     private readonly ServiceBusReceiver? _receiver;
@@ -36,8 +35,19 @@ private QueueHeartbeat(
         _runningTask = RunAsync(_cts.Token);
     }
 
-    public QueueHeartbeat(ServiceBusReceiver receiver, ILogger logger, string entityPath)
-        : this(receiver, ct => receiver.PeekMessageAsync(cancellationToken: ct), s_defaultInterval, logger, entityPath)
+    public QueueHeartbeat(ServiceBusReceiver receiver, TimeSpan autoDeleteOnIdle, ILogger logger, string entityPath)
+        : this(
+            receiver,
+            ct => receiver.PeekMessageAsync(cancellationToken: ct),
+            autoDeleteOnIdle >= AzureServiceBusReceiveEndpointConfiguration.TemporaryDefaults.MinimumAutoDeleteOnIdle
+                ? autoDeleteOnIdle / 2
+                : throw new ArgumentOutOfRangeException(
+                    nameof(autoDeleteOnIdle),
+                    autoDeleteOnIdle,
+                    "AutoDeleteOnIdle must be at least "
+                        + $"{AzureServiceBusReceiveEndpointConfiguration.TemporaryDefaults.MinimumAutoDeleteOnIdle}."),
+            logger,
+            entityPath)
     { }
 
     internal QueueHeartbeat(Func<CancellationToken, Task> peek, TimeSpan interval, ILogger logger, string entityPath)

src/Mocha/src/Mocha.Transport.AzureServiceBus/ThrowHelper.cs

@@ -85,6 +85,16 @@ public static Exception TopologyTopicNotFound(string topicName)
     public static Exception TopologyQueueNotFound(string queueName)
         => new InvalidOperationException($"Queue '{queueName}' not found in topology");
 
+    public static Exception TemporaryEndpointQueueAutoDeleteOnIdleConflict(
+        string endpointName,
+        string queueName,
+        TimeSpan? declaredAutoDeleteOnIdle)
+        => new InvalidOperationException(
+            $"Endpoint '{endpointName}' is marked Temporary(), but queue '{queueName}' was already declared "
+            + $"with AutoDeleteOnIdle '{declaredAutoDeleteOnIdle?.ToString() ?? "unset"}', which does not "
+            + "match the endpoint's resolved idle timeout. Align the queue declaration's AutoDeleteOnIdle "
+            + "with the endpoint's Temporary() idle timeout.");
+
     public static Exception SubscriptionAlreadyExists(string source, string destination)
         => new InvalidOperationException(
             $"Subscription from topic '{source}' to queue '{destination}' already exists");

src/Mocha/src/Mocha.Transport.AzureServiceBus/Topology/AzureServiceBusQueue.cs

@@ -120,6 +120,11 @@ internal void SetForwardDeadLetteredMessagesTo(string entityName)
         ForwardDeadLetteredMessagesTo = entityName;
     }
 
+    internal void SetAutoDeleteOnIdle(TimeSpan autoDeleteOnIdle)
+    {
+        AutoDeleteOnIdle = autoDeleteOnIdle;
+    }
+
     /// <inheritdoc />
     public async Task ProvisionAsync(
         AzureServiceBusClientManager clientManager,
@@ -192,4 +197,20 @@ public async Task ProvisionAsync(
             // Already provisioned by another instance, safe to ignore.
         }
     }
+
+    /// <summary>
+    /// Deletes this queue from Azure Service Bus.
+    /// </summary>
+    internal async Task DeprovisionAsync(
+        AzureServiceBusClientManager clientManager,
+        CancellationToken cancellationToken)
+    {
+        try
+        {
+            await clientManager.DeleteQueueAsync(Name, cancellationToken);
+        }
+        catch (ServiceBusException ex) when (ex.Reason == ServiceBusFailureReason.MessagingEntityNotFound)
+        {
+        }
+    }
 }