[Mocha] Configure receive resilience on PostgreSQL queues

src/Mocha/src/Mocha.Transport.Postgres/Configurations/PostgresQueueDescriptorConfiguration.cs

@@ -67,6 +67,11 @@ internal PostgresQueueDescriptorConfiguration(string name)
     /// </summary>
     public List<Action<List<ReceiveMiddlewareConfiguration>>> ReceivePipelineModifiers { get; } = [];
 
+    /// <summary>
+    /// Gets the configuration callbacks applied to this queue's receive endpoint, in registration order.
+    /// </summary>
+    public List<Action<IReceiveMiddlewareProvider>> ConfigureReceiveEndpoint { get; } = [];
+
     /// <summary>
     /// Gets the source topic bindings declared from this queue descriptor.
     /// </summary>

src/Mocha/src/Mocha.Transport.Postgres/Descriptors/IPostgresQueueDescriptor.cs

@@ -2,8 +2,11 @@ namespace Mocha.Transport.Postgres;
 
 /// <summary>
 /// Fluent descriptor for configuring a PostgreSQL queue and its receive endpoint.
+/// Receive configuration callbacks apply to the queue's receive endpoint.
 /// </summary>
-public interface IPostgresQueueDescriptor : IMessagingDescriptor<PostgresQueueDescriptorConfiguration>
+public interface IPostgresQueueDescriptor
+    : IMessagingDescriptor<PostgresQueueDescriptorConfiguration>
+    , IConfigurable<IReceiveMiddlewareProvider>
 {
     /// <summary>
     /// Sets whether the backing queue is automatically provisioned in the database.

src/Mocha/src/Mocha.Transport.Postgres/Descriptors/PostgresMessagingTransportDescriptor.cs

@@ -340,6 +340,11 @@ private void ConfigureQueueEndpoint(PostgresQueueDescriptorConfiguration configu
         target.ReceivePipelineModifiers.AddRange(configuration.ReceivePipelineModifiers);
         CopyFaultEndpointFeature(configuration, target);
         CopySkippedEndpointFeature(configuration, target);
+
+        foreach (var configure in configuration.ConfigureReceiveEndpoint)
+        {
+            configure(endpoint);
+        }
     }
 
     private static void CopyFaultEndpointFeature(

src/Mocha/src/Mocha.Transport.Postgres/Descriptors/PostgresQueueDescriptor.cs

@@ -145,6 +145,14 @@ public IPostgresQueueDescriptor UseReceive(
         return this;
     }
 
+    /// <inheritdoc />
+    public void Configure(Action<IReceiveMiddlewareProvider> configure)
+    {
+        ArgumentNullException.ThrowIfNull(configure);
+
+        Configuration.ConfigureReceiveEndpoint.Add(configure);
+    }
+
     /// <inheritdoc />
     public IPostgresQueueDescriptor FaultEndpoint(Uri address)
     {

src/Mocha/test/Mocha.Transport.Postgres.Tests/Descriptors/PostgresUnifiedQueueTests.cs

@@ -205,6 +205,66 @@ public void Queue_Should_PropagateAutoProvision_When_AutoProvisionCalled()
         Assert.Equal(false, queue.AutoProvision);
     }
 
+    [Fact]
+    public void Queue_Should_ConfigureEndpointReceiveFeatures_When_ResilienceConfiguredOnQueue()
+    {
+        // arrange
+        var runtime = CreateRuntime(
+            b => b.AddConsumer<OrderSpyConsumer>(),
+            t =>
+            {
+                t.BindExplicitly();
+                t.Queue("orders")
+                    .Consumer<OrderSpyConsumer>()
+                    .AddResilience(p => p.On<InvalidOperationException>().Discard())
+                    .AddCircuitBreaker(o => o.FailureRatio = 0.5)
+                    .AddConcurrencyLimiter(o => o.MaxConcurrency = 4);
+            });
+        var transport = runtime.Transports.OfType<PostgresMessagingTransport>().Single();
+
+        // act
+        var endpoint = transport.ReceiveEndpoints
+            .OfType<PostgresReceiveEndpoint>()
+            .Single(e => e.Queue.Name == "orders");
+
+        // assert
+        Type[] expectedRules = [typeof(Exception), typeof(InvalidOperationException)];
+        Assert.Equal(
+            expectedRules,
+            endpoint.Features.Get<ExceptionPolicyFeature>()?.Rules.Select(r => r.ExceptionType));
+        Assert.Equal(0.5, endpoint.Features.Get<CircuitBreakerFeature>()?.FailureRatio);
+        Assert.Equal(4, endpoint.Features.Get<ConcurrencyLimiterFeature>()?.MaxConcurrency);
+    }
+
+    [Fact]
+    public void Queue_Should_CombineExceptionPolicies_When_EndpointWithSameNameConfiguresResilience()
+    {
+        // arrange
+        var runtime = CreateRuntime(
+            b => b.AddConsumer<OrderSpyConsumer>(),
+            t =>
+            {
+                t.BindExplicitly();
+                t.Queue("orders")
+                    .Consumer<OrderSpyConsumer>()
+                    .AddResilience(p => p.On<InvalidOperationException>().Discard());
+                t.Endpoint("orders").AddResilience(p => p.On<TimeoutException>().Discard());
+            });
+        var transport = runtime.Transports.OfType<PostgresMessagingTransport>().Single();
+
+        // act
+        var endpoint = transport.ReceiveEndpoints
+            .OfType<PostgresReceiveEndpoint>()
+            .Single(e => e.Queue.Name == "orders");
+
+        // assert
+        // endpoint callbacks run first, queue callbacks replay onto the same feature afterwards
+        Type[] expectedRules = [typeof(TimeoutException), typeof(Exception), typeof(InvalidOperationException)];
+        Assert.Equal(
+            expectedRules,
+            endpoint.Features.Get<ExceptionPolicyFeature>()?.Rules.Select(r => r.ExceptionType));
+    }
+
     private static MessagingRuntime CreateRuntime(
         Action<IMessageBusHostBuilder> configureBuilder,
         Action<IPostgresMessagingTransportDescriptor> configureTransport)