ability to make a global latch on sending failure acks. Closes GH-1249

src/Testing/CoreTests/Runtime/MoveToErrorQueueTester.cs

@@ -33,6 +33,20 @@ await theLifecycle
             ;
     }
 
+    [Fact]
+    public async Task should_not_send_a_failure_ack_if_wolverine_options_latches_that()
+    {
+        theEnvelope.Destination = new Uri("local://foo");
+        theRuntime.Options.EnableAutomaticFailureAcks = false;
+
+        await theContinuation.ExecuteAsync(theLifecycle, theRuntime, DateTimeOffset.Now, null);
+
+        await theLifecycle
+                .DidNotReceive()
+                .SendFailureAcknowledgementAsync($"Moved message {theEnvelope.Id} to the Error Queue.\n{theException}")
+            ;
+    }
+
     [Fact]
     public async Task should_not_send_a_failure_ack_if_local()
     {

src/Testing/CoreTests/WolverineOptionsTests.cs

@@ -19,6 +19,12 @@ public class WolverineOptionsTests
 {
     private readonly WolverineOptions theSettings = new();
 
+    [Fact]
+    public void failure_acks_are_enabled_by_default()
+    {
+        new WolverineOptions().EnableAutomaticFailureAcks.ShouldBeTrue();
+    }
+
     [Fact]
     public void multiple_handler_behavior_is_classic_mode_by_default()
     {

src/Wolverine/ErrorHandling/MoveToErrorQueue.cs

@@ -31,7 +31,7 @@ public async ValueTask ExecuteAsync(IEnvelopeLifecycle lifecycle,
     {
         // TODO -- at some point, we need a more systematic way of doing this
         var scheme = lifecycle.Envelope.Destination.Scheme;
-        if (scheme != TransportConstants.Local && scheme != "external-table")
+        if (runtime.Options.EnableAutomaticFailureAcks && scheme != TransportConstants.Local && scheme != "external-table")
         {
             await lifecycle.SendFailureAcknowledgementAsync(
                 $"Moved message {lifecycle.Envelope!.Id} to the Error Queue.\n{Exception}");

src/Wolverine/Transports/Sending/InlineSendingAgent.cs

@@ -39,6 +39,10 @@ private async Task sendWithTracing(Envelope e, CancellationToken cancellationTok
             await _sender.SendAsync(e);
             _messageLogger.Sent(e);
         }
+        catch (NotSupportedException)
+        {
+            // ignore it
+        }
         finally
         {
             activity?.Stop();

src/Wolverine/Transports/Sending/SendingAgent.cs

@@ -235,6 +235,10 @@ private async Task sendWithExplicitHandlingAsync(Envelope envelope, Cancellation
 
             await MarkSuccessfulAsync(envelope);
         }
+        catch (NotSupportedException)
+        {
+            // Ignore it, most likely a failure ack that should not have been sent
+        }
         catch (Exception e)
         {
             try

src/Wolverine/WolverineOptions.cs

@@ -159,6 +159,12 @@ public MultipleHandlerBehavior MultipleHandlerBehavior
     /// </summary>
     public bool EnableRemoteInvocation { get; set; } = true;
 
+    /// <summary>
+    /// Should message failures automatically try to send a failure acknowledgement message back to the
+    /// original caller. Default is true.
+    /// </summary>
+    public bool EnableAutomaticFailureAcks { get; set; } = true;
+
     private void deriveServiceName()
     {
         if (GetType() == typeof(WolverineOptions))