Add sync way to retry in InlineSendingAgent

src/Wolverine/DurabilitySettings.cs

@@ -201,6 +201,12 @@ internal set
     /// </summary>
     public TimeSpan NodeEventRecordExpirationTime { get; set; } = 5.Days();
 
+    /// <summary>
+    /// When this option is enabled retry block used in InlineSendingAgent will synchronously wait on sending task to assure the message is send. 
+    /// When set to <see langword="false"/> default behavior is used so InlineSendingAgent agent will try to send a message and when failed it will give control to caller and retry on other thread in async manner
+    /// </summary>
+    public bool UseSyncRetryBlock { get; set; }
+
     /// <summary>
     ///     Get or set the logical Wolverine service name. By default, this is
     ///     derived from the name of a custom WolverineOptions

src/Wolverine/Transports/Sending/InlineSendingAgent.cs

@@ -9,7 +9,7 @@ namespace Wolverine.Transports.Sending;
 public class InlineSendingAgent : ISendingAgent, IDisposable
 {
     private readonly IMessageTracker _messageLogger;
-    private readonly RetryBlock<Envelope> _sending;
+    private readonly IRetryBlock<Envelope> _sending;
     private readonly DurabilitySettings _settings;
 
     public InlineSendingAgent(ILogger logger, ISender sender, Endpoint endpoint, IMessageTracker messageLogger,
@@ -20,16 +20,18 @@ public InlineSendingAgent(ILogger logger, ISender sender, Endpoint endpoint, IMe
         _settings = settings;
         Endpoint = endpoint;
 
-        if (endpoint.TelemetryEnabled)
+        if (settings.UseSyncRetryBlock)
         {
-            _sending = new RetryBlock<Envelope>(sendWithTracing, logger, _settings.Cancellation);
+            _sending = new RetryBlockSync<Envelope>(RetryHandlerResolver(endpoint), logger, _settings.Cancellation);
         }
         else
         {
-            _sending = new RetryBlock<Envelope>(sendWithOutTracing, logger, _settings.Cancellation);
+            _sending = new RetryBlock<Envelope>(RetryHandlerResolver(endpoint), logger, _settings.Cancellation);
         }
     }
 
+    private Func<Envelope, CancellationToken, Task> RetryHandlerResolver(Endpoint endpoint) => endpoint.TelemetryEnabled ? sendWithTracing : sendWithOutTracing;
+
     public ISender Sender { get; }
 
     private async Task sendWithTracing(Envelope e, CancellationToken cancellationToken)

src/Wolverine/Util/Dataflow/IRetryBlock.cs

@@ -0,0 +1,15 @@
+namespace Wolverine.Util.Dataflow;
+
+/// <summary>
+/// Abstract a way we can retry on <typeparamref name="T"/> message
+/// </summary>
+/// <typeparam name="T"></typeparam>
+internal interface IRetryBlock<T> : IDisposable
+{
+    /// <summary>
+    /// Send <typeparamref name="T"/> message in async manner
+    /// </summary>
+    /// <param name="message"></param>
+    /// <returns></returns>
+    Task PostAsync(T message);
+}

src/Wolverine/Util/Dataflow/RetryBlock.cs

@@ -1,6 +1,7 @@
-using System.Threading.Tasks.Dataflow;
+using JasperFx.Blocks;
 using JasperFx.Core;
 using Microsoft.Extensions.Logging;
+using System.Threading.Tasks.Dataflow;
 
 namespace Wolverine.Util.Dataflow;
 
@@ -24,7 +25,7 @@ public Task ExecuteAsync(T message, CancellationToken cancellation)
     }
 }
 
-public class RetryBlock<T> : IDisposable
+public class RetryBlock<T> : IRetryBlock<T>
 {
     private readonly ActionBlock<Item> _block;
     private readonly CancellationToken _cancellationToken;

src/Wolverine/Util/Dataflow/RetryBlockSync.cs

@@ -0,0 +1,143 @@
+using JasperFx.Blocks;
+using JasperFx.Core;
+using Microsoft.Extensions.Logging;
+using System.Threading.Tasks.Dataflow;
+
+namespace Wolverine.Util.Dataflow;
+
+public class RetryBlockSync<T> : IRetryBlock<T>
+{
+    private readonly ActionBlock<Item> _block;
+    private readonly CancellationToken _cancellationToken;
+    private readonly IItemHandler<T> _handler;
+    private readonly ILogger _logger;
+
+    public RetryBlockSync(Func<T, CancellationToken, Task> handler, ILogger logger, CancellationToken cancellationToken,
+        Action<ExecutionDataflowBlockOptions>? configure = null)
+        : this(new LambdaItemHandler<T>(handler), logger, cancellationToken, configure)
+    {
+    }
+
+    public RetryBlockSync(Func<T, CancellationToken, Task> handler, ILogger logger, CancellationToken cancellationToken,
+        ExecutionDataflowBlockOptions options)
+    {
+        _handler = new LambdaItemHandler<T>(handler);
+        _logger = logger;
+
+        options.CancellationToken = cancellationToken;
+        options.SingleProducerConstrained = true;
+
+        _cancellationToken = cancellationToken;
+
+        _block = new ActionBlock<Item>(executeAsync, options);
+    }
+
+    public RetryBlockSync(IItemHandler<T> handler, ILogger logger, CancellationToken cancellationToken,
+        Action<ExecutionDataflowBlockOptions>? configure = null)
+    {
+        _handler = handler;
+        _logger = logger;
+        var options = new ExecutionDataflowBlockOptions
+        {
+            CancellationToken = cancellationToken,
+            SingleProducerConstrained = true
+        };
+
+        configure?.Invoke(options);
+
+        _cancellationToken = cancellationToken;
+
+        _block = new ActionBlock<Item>(executeAsync, options);
+    }
+
+    public int MaximumAttempts { get; set; } = 4;
+    public TimeSpan[] Pauses { get; set; } = [50.Milliseconds(), 100.Milliseconds(), 250.Milliseconds()];
+
+    public void Dispose()
+    {
+        _block.Complete();
+    }
+
+    public void Post(T message)
+    {
+        if (_cancellationToken.IsCancellationRequested) return;
+
+        var item = new Item(message);
+        _block.Post(item);
+    }
+
+    public Task PostAsync(T message)
+    {
+        if (_cancellationToken.IsCancellationRequested) return Task.CompletedTask;
+
+        Post(message);
+
+        return _block.Completion;
+    }
+
+    public TimeSpan DeterminePauseTime(int attempt)
+    {
+        if (attempt >= Pauses.Length)
+        {
+            return Pauses.LastOrDefault();
+        }
+
+        return Pauses[attempt - 1];
+    }
+
+    private async Task executeAsync(Item item)
+    {
+        if (_cancellationToken.IsCancellationRequested) return;
+
+        try
+        {
+            item.Attempts++;
+
+            if (item.Attempts > 1)
+            {
+                var pause = DeterminePauseTime(item.Attempts);
+                await Task.Delay(pause, _cancellationToken);
+            }
+            await _handler.ExecuteAsync(item.Message, _cancellationToken);
+            _logger.LogDebug("Completed {Item}", item.Message);
+
+            _block.Complete();
+        }
+        catch (Exception e)
+        {
+            _logger.LogError(e, "Error while trying to retry {Item}", item.Message);
+
+            if (_cancellationToken.IsCancellationRequested) return;
+
+            if (item.Attempts < MaximumAttempts)
+            {
+                _block.Post(item);
+            }
+            else
+            {
+                _logger.LogInformation("Discarding message {Message} after {Attempts} attempts", item.Message,
+                    item.Attempts);
+
+                throw;
+            }
+        }
+    }
+
+    public Task DrainAsync()
+    {
+        _block.Complete();
+        return _block.Completion;
+    }
+
+    public class Item
+    {
+        public Item(T item)
+        {
+            Message = item;
+            Attempts = 0;
+        }
+
+        public int Attempts { get; set; }
+        public T Message { get; }
+    }
+}