Undoing a change on MessageContext that broke existing behavior

src/Testing/CoreTests/Runtime/MockWolverineRuntime.cs

@@ -54,6 +54,7 @@ public MockWolverineRuntime()
     {
         Tracker.Subscribe(this);
         MetricsAccumulator = new MetricsAccumulator(this);
+        Options.ServiceName = "Mock";
     }
 
     public MetricsAccumulator MetricsAccumulator { get; }

src/Transports/RabbitMQ/Wolverine.RabbitMQ.Tests/Bugs/Bug_1801_not_acking_on_consumer_failure.cs

@@ -0,0 +1,38 @@
+using JasperFx.Core;
+using Microsoft.Extensions.Hosting;
+using Wolverine.Tracking;
+using Xunit;
+
+namespace Wolverine.RabbitMQ.Tests.Bugs;
+
+public class Bug_1801_not_acking_on_consumer_failure
+{
+    [Fact]
+    public async Task try_it()
+    {
+        using var host = await Host.CreateDefaultBuilder()
+            .UseWolverine(opts =>
+            {
+                opts.UseRabbitMq().AutoProvision();
+
+                opts.ListenToRabbitQueue("will_error");
+                opts.PublishMessage<CauseError>().ToRabbitQueue("will_error");
+            }).StartAsync();
+
+        var tracked = await host.TrackActivity()
+            .IncludeExternalTransports()
+            .DoNotAssertOnExceptionsDetected()
+            .Timeout(1.Minutes())
+            .SendMessageAndWaitAsync(new CauseError("Bang!"));
+    }
+}
+
+public record CauseError(string Message);
+
+public static class CauseErrorHandler
+{
+    public static void Handle(CauseError msg)
+    {
+        throw new InvalidOperationException(msg.Message);
+    }
+}

src/Transports/RabbitMQ/Wolverine.RabbitMQ.Tests/Wolverine.RabbitMQ.Tests.csproj

@@ -17,6 +17,7 @@
     <ItemGroup>
         <ProjectReference Include="..\..\..\Extensions\Wolverine.FluentValidation\Wolverine.FluentValidation.csproj" />
         <ProjectReference Include="..\..\..\Persistence\Wolverine.Marten\Wolverine.Marten.csproj"/>
+        <ProjectReference Include="..\..\..\Testing\CoreTests\CoreTests.csproj" />
         <ProjectReference Include="..\Wolverine.RabbitMQ\Wolverine.RabbitMQ.csproj"/>
         <ProjectReference Include="..\..\..\Persistence\Wolverine.SqlServer\Wolverine.SqlServer.csproj" />
         <ProjectReference Include="..\..\..\Testing\Wolverine.ComplianceTests\Wolverine.ComplianceTests.csproj" />

src/Transports/RabbitMQ/Wolverine.RabbitMQ.Tests/channel_configuration.cs

@@ -23,7 +23,7 @@ public void can_customize_channel_creation()
 
         wolverineOptions.PublisherConfirmationsEnabled.ShouldBe(true);
         wolverineOptions.PublisherConfirmationTrackingEnabled.ShouldBe(true);
-        wolverineOptions.ConsumerDispatchConcurrency.ShouldBeEquivalentTo(5);
+        wolverineOptions.ConsumerDispatchConcurrency.ShouldBeEquivalentTo((ushort)5);
     }
 
     [Fact]
@@ -45,6 +45,6 @@ public void can_customize_channel_creation_additively()
 
         wolverineOptions.PublisherConfirmationsEnabled.ShouldBe(true);
         wolverineOptions.PublisherConfirmationTrackingEnabled.ShouldBe(false);
-        wolverineOptions.ConsumerDispatchConcurrency.ShouldBeEquivalentTo(2);
+        wolverineOptions.ConsumerDispatchConcurrency.ShouldBeEquivalentTo((ushort)2);
     }
 }

src/Transports/RabbitMQ/Wolverine.RabbitMQ.Tests/end_to_end.cs

@@ -69,7 +69,7 @@ public async Task rabbitmq_transport_is_exposed_as_a_resource()
             opts.Services.AddResourceSetupOnStartup(StartupAction.ResetState);
         });
 
-        var sources = publisher.Services.GetServices<ISystemPart>();
+        var sources = publisher.Services.GetServices<ISystemPart>().OfType<WolverineSystemPart>();
         foreach (var source in sources)
         {
             var resources = await source.FindResources();

src/Transports/RabbitMQ/Wolverine.RabbitMQ.Tests/exchange_queue_binding_model_setup_and_teardown_smoke_tests.cs

@@ -1,3 +1,4 @@
+using CoreTests.Runtime;
 using JasperFx.Core;
 using Microsoft.Extensions.Logging.Abstractions;
 using NSubstitute;
@@ -38,9 +39,7 @@ public exchange_queue_binding_model_setup_and_teardown_smoke_tests()
             .BindExchange("fan1")
             .ToQueue("xqueue2", "key2");
 
-        var wolverineRuntime = Substitute.For<IWolverineRuntime>();
-        wolverineRuntime.Logger.Returns(NullLogger.Instance);
-        wolverineRuntime.DurabilitySettings.Returns(new DurabilitySettings());
+        var wolverineRuntime = new MockWolverineRuntime();
         theTransport.TryBuildStatefulResource(wolverineRuntime, out var resource);
 
         theResource = resource;

src/Transports/RabbitMQ/Wolverine.RabbitMQ.Tests/exclusive_listeners.cs

@@ -62,7 +62,7 @@ public async Task exclusive_listeners_are_automatically_started_in_solo_mode()
             }).StartAsync();
 
         var runtime = host.GetRuntime();
-        runtime.Endpoints.ActiveListeners().Select(x => x.Endpoint.EndpointName)
+        runtime.Endpoints.ActiveListeners().Where(x => x.Uri.Scheme != "stub" ).Select(x => x.Endpoint.EndpointName)
             .OrderBy(x => x)
             .ShouldHaveTheSameElementsAs("one", "three", "two");
     }

src/Transports/RabbitMQ/Wolverine.RabbitMQ.Tests/leader_pinned_listener.cs

@@ -79,7 +79,7 @@ await host.WaitUntilAssignmentsChangeTo(w =>
             w.ExpectRunningAgents(host, 2);
         }, 30.Seconds());
 
-        var listeners = host.GetRuntime().Endpoints.ActiveListeners().Where(x => x.Endpoint.Role == EndpointRole.Application).Select(x => x.Uri).ToArray();
+        var listeners = host.GetRuntime().Endpoints.ActiveListeners().Where(x => x.Uri.Scheme != "stub" && x.Endpoint.Role == EndpointRole.Application).Select(x => x.Uri).ToArray();
         listeners.Length.ShouldBe(2);
         listeners.ShouldContain(new Uri("rabbitmq://queue/admin1"));
         listeners.ShouldContain(new Uri("rabbitmq://queue/admin2"));

src/Transports/RabbitMQ/Wolverine.RabbitMQ.Tests/multi_tenancy_through_virtual_hosts.cs

@@ -45,7 +45,7 @@ public async Task InitializeAsync()
 
                 opts.ServiceName = "main";
 
-                opts.UseRabbitMq().AutoProvision().AutoPurgeOnStartup()
+                opts.UseRabbitMq().AutoProvision().AutoPurgeOnStartup().DisableDeadLetterQueueing()
                     .AddTenant("one", "vh1")
                     .AddTenant("two", "vh2")
                     .AddTenant("three", "vh3");
@@ -78,7 +78,7 @@ public async Task InitializeAsync()
             {
                 opts.Policies.DisableConventionalLocalRouting();
                 opts.ServiceName = "one";
-                opts.UseRabbitMq(f => f.VirtualHost = "vh1");
+                opts.UseRabbitMq(f => f.VirtualHost = "vh1").DisableDeadLetterQueueing();
                 opts.ListenToRabbitQueue("multi_incoming");
 
                 opts.Services.AddResourceSetupOnStartup();
@@ -91,7 +91,7 @@ public async Task InitializeAsync()
             {
                 opts.Policies.DisableConventionalLocalRouting();
                 opts.ServiceName = "two";
-                opts.UseRabbitMq(f => f.VirtualHost = "vh2");
+                opts.UseRabbitMq(f => f.VirtualHost = "vh2").DisableDeadLetterQueueing();
                 opts.ListenToRabbitQueue("multi_incoming");
 
                 opts.Services.AddResourceSetupOnStartup();
@@ -102,7 +102,7 @@ public async Task InitializeAsync()
             {
                 opts.Policies.DisableConventionalLocalRouting();
                 opts.ServiceName = "three";
-                opts.UseRabbitMq(f => f.VirtualHost = "vh3");
+                opts.UseRabbitMq(f => f.VirtualHost = "vh3").DisableDeadLetterQueueing();
                 opts.ListenToRabbitQueue("multi_incoming");
 
                 opts.Services.AddResourceSetupOnStartup();
@@ -155,6 +155,7 @@ public async Task send_message_to_a_specific_tenant()
         var message = new MultiTenantMessage(Guid.NewGuid());
         var session = await _fixture.Main
             .TrackActivity()
+            .Timeout(15.Seconds())
             .AlsoTrack(_fixture.One, _fixture.Two, _fixture.Three)
             .WaitForMessageToBeReceivedAt<MultiTenantMessage>(_fixture.Two)
             .SendMessageAndWaitAsync(message, new DeliveryOptions{TenantId = "two"});

src/Transports/RabbitMQ/Wolverine.RabbitMQ/Internal/RabbitMqChannelCallback.cs

@@ -99,6 +99,8 @@ private async Task moveToErrorQueueAsync(RabbitMqEnvelope envelope, Cancellation
         {
             if (envelope.RabbitMqListener.Channel is not null)
             {
+                // For idempotency
+                envelope.HasBeenAcked = true;
                 await envelope.RabbitMqListener.Channel.BasicNackAsync(envelope.DeliveryTag, false, false, token);
             }
         }

src/Transports/RabbitMQ/Wolverine.RabbitMQ/Internal/RabbitMqExchange.cs

@@ -94,7 +94,7 @@ internal override string RoutingKey()
 
     internal async Task DeclareAsync(IChannel channel, ILogger logger)
     {
-        if (HasDeclared || DeclaredName == string.Empty)
+        if (DeclaredName == string.Empty)
         {
             return;
         }

src/Wolverine/Envelope.Internals.cs

@@ -84,6 +84,8 @@ internal Envelope(object message, IMessageSerializer writer)
     public bool IsResponse { get; set; }
     public Exception? Failure { get; set; }
     internal Envelope[]? Batch { get; set; }
+
+    internal bool HasBeenAcked { get; set; }
 
     internal void StartTiming()
     {

src/Wolverine/ErrorHandling/MoveToErrorQueue.cs

@@ -43,6 +43,8 @@ await lifecycle.SendFailureAcknowledgementAsync(
         }
 
         await lifecycle.MoveToDeadLetterQueueAsync(Exception);
+
+        await lifecycle.CompleteAsync();
 
         activity?.AddEvent(new ActivityEvent(WolverineTracing.MovedToErrorQueue));
 

src/Wolverine/Runtime/MessageContext.cs

@@ -184,25 +184,28 @@ public async Task AssertAnyRequiredResponseWasGenerated()
         }
     }
 
-    public ValueTask CompleteAsync()
+    public async ValueTask CompleteAsync()
     {
         if (_channel == null || Envelope == null)
         {
             throw new InvalidOperationException("No Envelope is active for this context");
         }
 
-        return _channel.CompleteAsync(Envelope);
+        if (Envelope.HasBeenAcked) return;
+
+        await _channel.CompleteAsync(Envelope);
+        Envelope.HasBeenAcked = true;
     }
 
-    public ValueTask DeferAsync()
+    public async ValueTask DeferAsync()
     {
         if (_channel == null || Envelope == null)
         {
             throw new InvalidOperationException("No Envelope is active for this context");
         }
 
         Runtime.MessageTracking.Requeued(Envelope);
-        return _channel.DeferAsync(Envelope);
+        await _channel.DeferAsync(Envelope);
     }
 
     public async Task ReScheduleAsync(DateTimeOffset scheduledTime)