Declare Mocha's temporary RabbitMQ queues durable with auto-delete and expiry

src/CookieCrumble/src/CookieCrumble.Resources.RabbitMQ/RabbitMQResource.cs

@@ -27,7 +27,7 @@ public async Task<string> InvokeCommandAsync(IEnumerable<string> command)
 
     protected override RabbitMqContainer Build()
         => Configure(
-            new RabbitMqBuilder("rabbitmq:3.11")
+            new RabbitMqBuilder("rabbitmq:4.3")
                 .WithUsername("guest")
                 .WithPassword("guest"))
             .Build();

src/Mocha/src/Mocha.Transport.RabbitMQ/Configurations/RabbitMQQueueDescriptorConfiguration.cs

@@ -61,6 +61,12 @@ internal RabbitMQQueueDescriptorConfiguration(string name)
     /// </summary>
     public bool IsTemporary { get; set; }
 
+    /// <summary>
+    /// Gets or sets the queue expiry applied when this queue's receive endpoint is temporary. When
+    /// <see langword="null"/>, <see cref="RabbitMQReceiveEndpointConfiguration.TemporaryDefaults.Expiry"/> applies.
+    /// </summary>
+    public TimeSpan? TemporaryExpiry { get; set; }
+
     /// <summary>
     /// Gets the receive middleware configurations applied when this queue materializes an endpoint.
     /// </summary>

src/Mocha/src/Mocha.Transport.RabbitMQ/Configurations/RabbitMQReceiveEndpointConfiguration.cs

@@ -15,4 +15,24 @@ public sealed class RabbitMQReceiveEndpointConfiguration : ReceiveEndpointConfig
     /// Defaults to 100.
     /// </summary>
     public ushort MaxPrefetch { get; set; } = 100;
+
+    /// <summary>
+    /// Gets or sets the queue expiry applied to a temporary endpoint's queue. When <see langword="null"/>
+    /// and <see cref="ReceiveEndpointConfiguration.IsTemporary"/> is set, <see cref="TemporaryDefaults.Expiry"/> applies.
+    /// </summary>
+    public TimeSpan? TemporaryExpiry { get; set; }
+
+    public static class TemporaryDefaults
+    {
+        /// <summary>
+        /// The queue expiry applied to a temporary endpoint marked via the parameterless
+        /// <c>Temporary()</c> overload.
+        /// </summary>
+        public static readonly TimeSpan Expiry = TimeSpan.FromMinutes(30);
+
+        /// <summary>
+        /// The longest queue expiry the transport can declare through <c>x-expires</c>.
+        /// </summary>
+        public static readonly TimeSpan MaximumExpiry = TimeSpan.FromMilliseconds(int.MaxValue);
+    }
 }

src/Mocha/src/Mocha.Transport.RabbitMQ/Descriptors/IRabbitMQQueueDescriptor.cs

@@ -112,12 +112,23 @@ public interface IRabbitMQQueueDescriptor : IMessagingDescriptor<RabbitMQQueueDe
     IRabbitMQQueueDescriptor MaxConcurrency(int maxConcurrency);
 
     /// <summary>
-    /// Marks this queue's receive endpoint as temporary, provisioning a non-durable, auto-delete
-    /// queue whose lifecycle is scoped to the lifetime of the consuming process.
+    /// Marks this queue's receive endpoint as temporary, provisioning a durable, auto-delete queue
+    /// with the default queue expiry.
     /// </summary>
     /// <returns>The descriptor for method chaining.</returns>
     IRabbitMQQueueDescriptor Temporary();
 
+    /// <summary>
+    /// Marks this queue's receive endpoint as temporary with an explicit queue expiry, after which
+    /// the broker removes the queue once it has no consumers.
+    /// </summary>
+    /// <param name="expiry">
+    /// The queue expiry. Must be positive and at most
+    /// <see cref="RabbitMQReceiveEndpointConfiguration.TemporaryDefaults.MaximumExpiry"/>.
+    /// </param>
+    /// <returns>The descriptor for method chaining.</returns>
+    IRabbitMQQueueDescriptor Temporary(TimeSpan expiry);
+
     /// <summary>
     /// Adds receive middleware to this queue's endpoint pipeline.
     /// </summary>

src/Mocha/src/Mocha.Transport.RabbitMQ/Descriptors/IRabbitMQReceiveEndpointDescriptor.cs

@@ -39,6 +39,17 @@ public interface IRabbitMQReceiveEndpointDescriptor : IReceiveEndpointDescriptor
     /// <inheritdoc cref="IReceiveEndpointDescriptor{TConfiguration}.Temporary" />
     new IRabbitMQReceiveEndpointDescriptor Temporary();
 
+    /// <summary>
+    /// Marks this receive endpoint as temporary with an explicit queue expiry, after which the
+    /// broker removes the queue once it has no consumers.
+    /// </summary>
+    /// <param name="expiry">
+    /// The queue expiry. Must be positive and at most
+    /// <see cref="RabbitMQReceiveEndpointConfiguration.TemporaryDefaults.MaximumExpiry"/>.
+    /// </param>
+    /// <returns>The descriptor for method chaining.</returns>
+    IRabbitMQReceiveEndpointDescriptor Temporary(TimeSpan expiry);
+
     /// <summary>
     /// Sets the address of the fault endpoint where failed messages are forwarded.
     /// </summary>

src/Mocha/src/Mocha.Transport.RabbitMQ/Descriptors/RabbitMQMessagingTransportDescriptor.cs

@@ -268,6 +268,13 @@ private void ConfigureQueueTopology(RabbitMQQueueDescriptorConfiguration configu
         var queue = DeclareQueue(configuration.Name!);
         ApplyQueueConfiguration(configuration.Queue, queue);
 
+        if (configuration.IsTemporary && configuration.Queue.Arguments?.ContainsKey("x-expires") is not true)
+        {
+            queue.Expires(
+                configuration.TemporaryExpiry
+                ?? RabbitMQReceiveEndpointConfiguration.TemporaryDefaults.Expiry);
+        }
+
         var schema = Configuration.Schema ?? RabbitMQTransportConfiguration.DefaultSchema;
         foreach (var binding in configuration.SourceBindings)
         {
@@ -318,6 +325,7 @@ private void ConfigureQueueEndpoint(RabbitMQQueueDescriptorConfiguration configu
         if (configuration.IsTemporary)
         {
             target.IsTemporary = true;
+            target.TemporaryExpiry ??= configuration.TemporaryExpiry;
         }
 
         target.ReceiveMiddlewares.AddRange(configuration.ReceiveMiddlewares);

src/Mocha/src/Mocha.Transport.RabbitMQ/Descriptors/RabbitMQQueueDescriptor.cs

@@ -130,7 +130,20 @@ public IRabbitMQQueueDescriptor MaxConcurrency(int maxConcurrency)
     public IRabbitMQQueueDescriptor Temporary()
     {
         Configuration.IsTemporary = true;
-        Configuration.Queue.Durable = false;
+        Configuration.Queue.AutoDelete = true;
+        return this;
+    }
+
+    /// <inheritdoc />
+    public IRabbitMQQueueDescriptor Temporary(TimeSpan expiry)
+    {
+        ArgumentOutOfRangeException.ThrowIfLessThanOrEqual(expiry, TimeSpan.Zero);
+        ArgumentOutOfRangeException.ThrowIfGreaterThan(
+            expiry,
+            RabbitMQReceiveEndpointConfiguration.TemporaryDefaults.MaximumExpiry);
+
+        Configuration.IsTemporary = true;
+        Configuration.TemporaryExpiry = expiry;
         Configuration.Queue.AutoDelete = true;
         return this;
     }

src/Mocha/src/Mocha.Transport.RabbitMQ/Descriptors/RabbitMQReceiveEndpointDescriptor.cs

@@ -104,6 +104,20 @@ private RabbitMQReceiveEndpointDescriptor(IMessagingConfigurationContext discove
         return this;
     }
 
+    /// <inheritdoc />
+    public IRabbitMQReceiveEndpointDescriptor Temporary(TimeSpan expiry)
+    {
+        ArgumentOutOfRangeException.ThrowIfLessThanOrEqual(expiry, TimeSpan.Zero);
+        ArgumentOutOfRangeException.ThrowIfGreaterThan(
+            expiry,
+            RabbitMQReceiveEndpointConfiguration.TemporaryDefaults.MaximumExpiry);
+
+        base.Temporary();
+        Configuration.TemporaryExpiry = expiry;
+
+        return this;
+    }
+
     /// <inheritdoc />
     public IRabbitMQReceiveEndpointDescriptor FaultEndpoint(Uri address)
     {

src/Mocha/src/Mocha.Transport.RabbitMQ/ThrowHelper.cs

@@ -0,0 +1,13 @@
+namespace Mocha.Transport.RabbitMQ;
+
+internal static class ThrowHelper
+{
+    public static Exception BeforeAndAfterConflict()
+        => Mocha.ThrowHelper.BeforeAndAfterConflict();
+
+    public static Exception TemporaryEndpointQueueConflict(string queueName, string? endpointName)
+        => new InvalidOperationException(
+            $"Queue '{queueName}' is explicitly declared without auto-delete, which conflicts with "
+            + $"receive endpoint '{endpointName}' being marked Temporary(). Declare the queue with "
+            + "AutoDelete(), or remove Temporary() from the endpoint.");
+}

src/Mocha/src/Mocha.Transport.RabbitMQ/Topology/RabbitMQQueue.cs

@@ -75,12 +75,13 @@ internal void AddBinding(RabbitMQBinding binding)
     }
 
     /// <summary>
-    /// Marks this queue as non-durable and auto-delete.
+    /// Marks this queue as auto-delete and sets the queue expiry after which the broker removes it
+    /// when it has no consumers.
     /// </summary>
-    internal void MarkTemporary()
+    internal void MarkTemporary(TimeSpan expiry)
     {
-        Durable = false;
         AutoDelete = true;
+        Arguments = Arguments.SetItem("x-expires", (int)expiry.TotalMilliseconds);
     }
 
     /// <summary>

src/Mocha/src/Mocha.Transport.RabbitMQ/Topology/RabbitMQRoutingStrategy.cs

@@ -210,22 +210,22 @@ public override void DiscoverTopology(
 
         if (rabbitConfiguration.IsTemporary)
         {
-            EnsureNoDurableQueueConflict(rabbitConfiguration);
+            EnsureNoTemporaryQueueConflict(rabbitConfiguration);
         }
 
         var queue = _topology.GetOrAddQueue(
             rabbitConfiguration.QueueName,
             _ => new RabbitMQQueueConfiguration
             {
-                Durable = rabbitConfiguration.IsTemporary ? false : null,
                 AutoDelete = rabbitConfiguration.IsTemporary,
                 AutoProvision = rabbitConfiguration.AutoProvision,
                 Origin = TopologyOrigin.Endpoint
             });
 
         if (rabbitConfiguration.IsTemporary && queue.Origin is not TopologyOrigin.Declared)
         {
-            queue.MarkTemporary();
+            queue.MarkTemporary(rabbitConfiguration.TemporaryExpiry ??
+                RabbitMQReceiveEndpointConfiguration.TemporaryDefaults.Expiry);
         }
 
         if (rabbitEndpoint.Kind == ReceiveEndpointKind.Default)
@@ -465,21 +465,17 @@ private void ConfigureFaultOrSkippedEndpoint(
         }
     }
 
-    private void EnsureNoDurableQueueConflict(RabbitMQReceiveEndpointConfiguration rabbitConfiguration)
+    private void EnsureNoTemporaryQueueConflict(RabbitMQReceiveEndpointConfiguration rabbitConfiguration)
     {
         var existingQueue = _topology.Queues.FirstOrDefault(q => q.Name == rabbitConfiguration.QueueName);
         if (existingQueue is not { Origin: TopologyOrigin.Declared })
         {
             return;
         }
 
-        if (existingQueue.Durable || !existingQueue.AutoDelete)
+        if (!existingQueue.AutoDelete)
         {
-            throw new InvalidOperationException(
-                $"Queue '{rabbitConfiguration.QueueName}' is explicitly declared as durable or "
-                + "non-auto-delete, which conflicts with receive endpoint "
-                + $"'{rabbitConfiguration.Name}' being marked Temporary(). Declare the queue as "
-                + "non-durable and auto-delete, or remove Temporary() from the endpoint.");
+            throw ThrowHelper.TemporaryEndpointQueueConflict(existingQueue.Name, rabbitConfiguration.Name);
         }
     }
 

src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/Behaviors/RequestReplyTests.cs

@@ -1,3 +1,4 @@
+using CookieCrumble;
 using Microsoft.Extensions.DependencyInjection;
 using Mocha.Transport.RabbitMQ.Tests.Helpers;
 
@@ -129,6 +130,57 @@ public async Task RequestAsync_Should_ReturnCorrectResponse_When_MultipleRequest
         Assert.Equal("InTransit", shipmentResponse.Status);
     }
 
+    [Fact]
+    public async Task Transport_Should_DeclareDurableAutoDeleteReplyQueue_When_Started()
+    {
+        // arrange
+        // The broker denies non-durable, non-exclusive queues by default since RabbitMQ 4.3, so a
+        // host that starts at all proves the reply queue shape is accepted. The listing pins it.
+        await using var vhost = await _fixture.CreateVhostAsync();
+        await using var bus = await new ServiceCollection()
+            .AddSingleton(vhost.ConnectionFactory)
+            .AddMessageBus()
+            .AddRequestHandler<GetOrderStatusHandler>()
+            .AddRabbitMQ()
+            .BuildTestBusAsync();
+
+        // act
+        var output = await _fixture.InvokeCommandAsync(
+            [
+                "rabbitmqctl",
+                "list_queues",
+                "name",
+                "durable",
+                "auto_delete",
+                "exclusive",
+                "arguments",
+                "-p",
+                vhost.VhostName,
+                "--no-table-headers"
+            ]);
+        var replyQueue = output
+            .Split('\n', StringSplitOptions.RemoveEmptyEntries | StringSplitOptions.TrimEntries)
+            .Single(line => line.StartsWith("response-", StringComparison.Ordinal))
+            .Split('\t');
+
+        // assert
+        new
+        {
+            Durable = replyQueue[1],
+            AutoDelete = replyQueue[2],
+            Exclusive = replyQueue[3],
+            Arguments = replyQueue[4]
+        }.MatchInlineSnapshot(
+            """
+            {
+              "Durable": "true",
+              "AutoDelete": "true",
+              "Exclusive": "false",
+              "Arguments": "[{\"x-expires\",1800000},{\"x-queue-type\",\"classic\"}]"
+            }
+            """);
+    }
+
     public sealed class ProcessPaymentHandler(MessageRecorder recorder) : IEventRequestHandler<ProcessPayment>
     {
         public ValueTask HandleAsync(ProcessPayment request, CancellationToken cancellationToken)

src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/Descriptors/RabbitMQUnifiedQueueTests.cs

@@ -297,11 +297,11 @@ public void Queue_Should_ConfigureFaultEndpointViaQueueDescriptor_When_FaultEndp
     }
 
     [Fact]
-    public void Queue_Should_MaterializeNonDurableAutoDeleteQueue_When_TemporaryCalled()
+    public void Queue_Should_MaterializeDurableAutoDeleteQueueWithExpiry_When_TemporaryCalled()
     {
         // arrange
-        // Temporary() on the Queue() descriptor must produce a non-durable, auto-delete queue,
-        // matching the lifecycle contract of a directly-declared temporary receive endpoint.
+        // Temporary() on the Queue() descriptor must produce the same durable, auto-delete queue
+        // with a queue expiry as a temporary receive endpoint.
         var runtime = CreateRuntime(
             b => { },
             t =>
@@ -313,12 +313,51 @@ public void Queue_Should_MaterializeNonDurableAutoDeleteQueue_When_TemporaryCall
         var topology = (RabbitMQMessagingTopology)transport.Topology;
 
         // act
-        var queue = topology.Queues.SingleOrDefault(q => q.Name == "temp-audit");
+        var queue = topology.Queues.Single(q => q.Name == "temp-audit");
 
         // assert
-        Assert.NotNull(queue);
-        Assert.False(queue.Durable);
-        Assert.True(queue.AutoDelete);
+        new { queue.Durable, queue.Exclusive, queue.AutoDelete, queue.Arguments }.MatchInlineSnapshot(
+            """
+            {
+              "Durable": true,
+              "Exclusive": false,
+              "AutoDelete": true,
+              "Arguments": {
+                "x-expires": 1800000
+              }
+            }
+            """);
+    }
+
+    [Fact]
+    public void Queue_Should_UseConfiguredExpiry_When_TemporaryCalledWithExpiry()
+    {
+        // arrange
+        var runtime = CreateRuntime(
+            b => { },
+            t =>
+            {
+                t.BindExplicitly();
+                t.Queue("temp-audit").Temporary(TimeSpan.FromMinutes(5));
+            });
+        var transport = runtime.Transports.OfType<RabbitMQMessagingTransport>().Single();
+        var topology = (RabbitMQMessagingTopology)transport.Topology;
+
+        // act
+        var queue = topology.Queues.Single(q => q.Name == "temp-audit");
+
+        // assert
+        new { queue.Durable, queue.Exclusive, queue.AutoDelete, queue.Arguments }.MatchInlineSnapshot(
+            """
+            {
+              "Durable": true,
+              "Exclusive": false,
+              "AutoDelete": true,
+              "Arguments": {
+                "x-expires": 300000
+              }
+            }
+            """);
     }
 
     private static MessagingRuntime CreateRuntime(