Keep RabbitMQ retries local and honor dead-lettering on exhaustion

Foundatio.RabbitMQ.slnx

@@ -14,6 +14,7 @@
   <Folder Name="/Tests/">
     <Project Path="tests/Foundatio.RabbitMQ.Tests/Foundatio.RabbitMQ.Tests.csproj" />
     <Project Path="tests/Foundatio.RabbitMQ.AppHost/Foundatio.RabbitMQ.AppHost.csproj" />
+    <Project Path="tests/Foundatio.RabbitMQ.ProcessProbe/Foundatio.RabbitMQ.ProcessProbe.csproj" />
   </Folder>
   <Project Path="src/Foundatio.RabbitMQ/Foundatio.RabbitMQ.csproj" />
 </Solution>

src/Foundatio.RabbitMQ/Messaging/DeadLetterStrategy.cs

@@ -18,7 +18,9 @@ public enum DeadLetterStrategy
     AtMostOnce,
 
     /// <summary>
-    /// Guarantees message transfer to the dead-letter exchange using internal publisher confirms.
+    /// Retains broker-managed dead letters until target queues confirm acceptance; duplicates remain possible.
+    /// Applies only to source quorum queues, not provider-managed terminal handoffs.
+    /// Destination durability and availability remain separate requirements.
     /// Requires overflow to be set to reject-publish (not drop-head).
     /// Uses more memory and CPU. Only enable when dead-lettered messages must not be lost.
     /// </summary>

src/Foundatio.RabbitMQ/Messaging/RabbitMQMessageConverter.cs

@@ -0,0 +1,96 @@
+using System;
+using System.Collections.Generic;
+using System.Globalization;
+using System.Text;
+using Foundatio.Utility;
+using RabbitMQ.Client;
+using RabbitMQ.Client.Events;
+
+namespace Foundatio.Messaging;
+
+internal static class RabbitMQMessageConverter
+{
+    private const string PriorityPropertyKey = "Priority";
+
+    internal static IMessage Convert(BasicDeliverEventArgs envelope, Type? type, Func<IMessage, object?> deserialize)
+    {
+        // A handler can outlive the transport callback after cancellation, so own the payload.
+        var message = new Message(envelope.Body.ToArray(), deserialize)
+        {
+            Type = envelope.BasicProperties.Type,
+            ClrType = type,
+            CorrelationId = envelope.BasicProperties.CorrelationId,
+            UniqueId = envelope.BasicProperties.MessageId
+        };
+        if (envelope.BasicProperties.Headers is not null)
+            foreach (var header in envelope.BasicProperties.Headers)
+            {
+                if (header.Value is byte[] bytes)
+                    message.Properties[header.Key] = Encoding.UTF8.GetString(bytes);
+                else if (header.Value is not null && System.Convert.ToString(header.Value, CultureInfo.InvariantCulture) is { } value)
+                    message.Properties[header.Key] = value;
+            }
+        return message;
+    }
+
+    internal static BasicProperties CreateProperties(string messageType, MessageOptions options, RabbitMQMessageBusOptions settings)
+    {
+        bool delayed = options.DeliveryDelay.GetValueOrDefault() > TimeSpan.Zero;
+        var properties = new BasicProperties
+        {
+            MessageId = options.UniqueId ?? Guid.NewGuid().ToString("N"),
+            CorrelationId = options.CorrelationId,
+            Type = messageType,
+            Persistent = settings.IsDurable
+        };
+        if (settings.DefaultMessageTimeToLive.HasValue)
+            properties.Expiration = settings.DefaultMessageTimeToLive.Value.TotalMilliseconds.ToString(CultureInfo.InvariantCulture);
+        if (options.Properties.TryGetValue(PriorityPropertyKey, out string? priorityValue) && Byte.TryParse(priorityValue, NumberStyles.Integer, CultureInfo.InvariantCulture, out byte priority))
+            properties.Priority = priority;
+        if (options.Properties.Count > 0)
+        {
+            properties.Headers = new Dictionary<string, object?>();
+            foreach (var property in options.Properties)
+                if (!String.Equals(property.Key, PriorityPropertyKey, StringComparison.Ordinal))
+                    properties.Headers.Add(property.Key, property.Value);
+        }
+        if (delayed)
+        {
+            double delayMs = options.DeliveryDelay!.Value.TotalMilliseconds;
+            if (delayMs > Int32.MaxValue)
+                throw new ArgumentOutOfRangeException(nameof(options), "DeliveryDelay exceeds the supported delayed-exchange range.");
+            properties.Headers ??= new Dictionary<string, object?>();
+            properties.Headers[RabbitMQConstants.XDelayHeader] = (int)delayMs;
+        }
+        return properties;
+    }
+
+    internal static BasicProperties CopyHandoffProperties(BasicDeliverEventArgs envelope)
+    {
+        var properties = new BasicProperties(envelope.BasicProperties)
+        {
+            UserId = null,
+            Headers = envelope.BasicProperties.Headers is { } headers
+                ? new Dictionary<string, object?>(headers) : new Dictionary<string, object?>()
+        };
+        properties.Headers.Remove(RabbitMQConstants.CarbonCopyHeader);
+        properties.Headers.Remove(RabbitMQConstants.BlindCarbonCopyHeader);
+        properties.Headers.Remove(RabbitMQConstants.XDelayHeader);
+        if (!properties.Headers.ContainsKey(RabbitMQConstants.XOriginalMessageIdHeader))
+            properties.Headers[RabbitMQConstants.XOriginalMessageIdHeader] = properties.MessageId;
+        return properties;
+    }
+
+    internal static long GetRetryCountFromHeader(BasicDeliverEventArgs envelope)
+    {
+        if (envelope.BasicProperties.Headers?.TryGetValue(RabbitMQConstants.XDeliveryCountHeader, out object? value) is not true)
+            return 0;
+        if (value is long number)
+            return number >= 0 ? number : Int64.MaxValue;
+        string? text = value is byte[] bytes ? Encoding.UTF8.GetString(bytes) : System.Convert.ToString(value, CultureInfo.InvariantCulture);
+        // Invalid/overflowing metadata must not reset an exhausted budget.
+        return Int64.TryParse(text, NumberStyles.None, CultureInfo.InvariantCulture, out long count) && count >= 0
+            ? count : Int64.MaxValue;
+    }
+
+}

src/Foundatio.RabbitMQ/Messaging/RabbitMQTopology.cs

@@ -0,0 +1,175 @@
+using System;
+using System.Collections.Generic;
+using System.Text;
+using System.Threading;
+using System.Threading.Tasks;
+using Foundatio.Extensions;
+using Foundatio.Utility;
+using Microsoft.Extensions.Logging;
+using RabbitMQ.Client;
+using RabbitMQ.Client.Exceptions;
+
+namespace Foundatio.Messaging;
+
+internal sealed class RabbitMQTopology
+{
+    private static readonly Version _delayedExchangePluginIncompatibleVersion = new(4, 3);
+    private static readonly Version _globalQosRemovedVersion = new(4, 3);
+    private readonly RabbitMQMessageBusOptions _options;
+    private readonly ILogger _logger;
+    private bool? _delayedExchangePluginEnabled;
+    private readonly bool _isQuorumQueue;
+
+    internal bool IsQuorumQueue => _isQuorumQueue;
+    internal bool SupportsDelayedExchange => _delayedExchangePluginEnabled is true;
+
+    internal RabbitMQTopology(RabbitMQMessageBusOptions options, ILogger logger)
+    {
+        _options = options;
+        _logger = logger;
+        _isQuorumQueue = RabbitMQMessageBusOptions.IsQuorumQueue(options.Arguments);
+    }
+
+    internal Version? DetectServerVersion(IConnection connection)
+    {
+        var version = ParseServerVersion(connection.ServerProperties);
+        if (version is not null)
+            _logger.LogDebug("Connected to RabbitMQ server version {ServerVersion}", version);
+        return version;
+    }
+
+    /// <summary>Parse the broker version from the UTF-8 AMQP server property.</summary>
+    internal static Version? ParseServerVersion(IDictionary<string, object?>? serverProperties)
+    {
+        if (serverProperties?.TryGetValue("version", out var value) is not true || value is not byte[] bytes)
+            return null;
+        return Version.TryParse(Encoding.UTF8.GetString(bytes), out var version) ? version : null;
+    }
+
+    internal async Task<bool?> CreateDelayedExchangeAsync(IChannel channel, Version? serverVersion, CancellationToken cancellationToken)
+    {
+        // RabbitMQ 4.3 removed the Mnesia store required by the archived plugin.
+        // Probe earlier brokers (and unknown versions) to retain 4.2 delayed delivery.
+        if (serverVersion is not null && serverVersion >= _delayedExchangePluginIncompatibleVersion)
+        {
+            if (_delayedExchangePluginEnabled is not false)
+                _logger.LogWarning("The delayed-message exchange plugin is incompatible with RabbitMQ {ServerVersion}; delayed publications use in-process scheduling unless broker delivery is required", serverVersion);
+            _delayedExchangePluginEnabled = false;
+            return null;
+        }
+        try
+        {
+            await channel.ExchangeDeclareAsync(_options.Topic, "x-delayed-message", _options.IsDurable, false,
+                new Dictionary<string, object?> { [RabbitMQConstants.DelayedExchangeTypeArgument] = ExchangeType.Fanout }, cancellationToken: cancellationToken).AnyContext();
+            if (_delayedExchangePluginEnabled is not true)
+                _logger.LogWarning("The delayed-message exchange plugin is archived and incompatible with RabbitMQ 4.3+; review its performance and migration guidance before upgrading the broker");
+            _delayedExchangePluginEnabled = true;
+            return true;
+        }
+        catch (OperationInterruptedException exception) when (IsRegularExchangeFallback(exception))
+        {
+            // Unsupported plugin type, or an existing regular fanout topic, is a known
+            // non-scheduling topology. Other declaration/permission/network errors propagate.
+            _delayedExchangePluginEnabled = false;
+            return false;
+        }
+    }
+
+    private static bool IsRegularExchangeFallback(OperationInterruptedException exception)
+    {
+        var reason = exception.ShutdownReason;
+        if (reason is null || reason.ReplyCode is not (406 or 503))
+            return false;
+        return reason.ReplyText.Contains("unknown exchange type 'x-delayed-message'", StringComparison.OrdinalIgnoreCase)
+            || (reason.ReplyCode == 406
+                && reason.ReplyText.Contains("inequivalent arg 'type'", StringComparison.OrdinalIgnoreCase)
+                && reason.ReplyText.Contains("current is 'fanout'", StringComparison.OrdinalIgnoreCase));
+    }
+
+    internal Task DeclareSubscriptionExchangeAsync(IChannel channel, CancellationToken cancellationToken) =>
+        _delayedExchangePluginEnabled is true
+            ? channel.ExchangeDeclareAsync(_options.Topic, "x-delayed-message", _options.IsDurable, false,
+                new Dictionary<string, object?> { [RabbitMQConstants.DelayedExchangeTypeArgument] = ExchangeType.Fanout }, cancellationToken: cancellationToken)
+            : CreateRegularExchangeAsync(channel, cancellationToken);
+
+    internal Task CreateRegularExchangeAsync(IChannel channel, CancellationToken cancellationToken) =>
+        channel.ExchangeDeclareAsync(_options.Topic, ExchangeType.Fanout, _options.IsDurable, false, cancellationToken: cancellationToken);
+
+    internal async Task<string> CreateQueueAsync(IChannel channel, Version? serverVersion, CancellationToken cancellationToken)
+    {
+        var arguments = _options.Arguments is not null
+            ? new Dictionary<string, object?>(_options.Arguments) : new Dictionary<string, object?>();
+        bool isQuorumQueue = RabbitMQMessageBusOptions.IsQuorumQueue(arguments);
+        if (isQuorumQueue != _isQuorumQueue)
+            throw new InvalidOperationException("Queue type cannot change after the message bus is constructed.");
+        if (isQuorumQueue && _options.MaxPriority.HasValue)
+            throw new InvalidOperationException("MaxPriority applies only to classic queues and cannot be used with quorum queues.");
+
+        if (_isQuorumQueue && !arguments.ContainsKey(RabbitMQConstants.DeliveryLimitArgument))
+            arguments[RabbitMQConstants.DeliveryLimitArgument] = _options.DeliveryLimit;
+        if (!String.IsNullOrWhiteSpace(_options.DeadLetterExchange))
+        {
+            arguments[RabbitMQConstants.DeadLetterExchangeArgument] = _options.DeadLetterExchange;
+            if (_options.DeadLetterRoutingKey is not null)
+                arguments[RabbitMQConstants.DeadLetterRoutingKeyArgument] = _options.DeadLetterRoutingKey;
+            if (_options.DeadLetterStrategy.HasValue)
+            {
+                if (_options.DeadLetterStrategy == DeadLetterStrategy.AtLeastOnce)
+                {
+                    if (!_isQuorumQueue)
+                        throw new MessageBusException("At-least-once broker dead-lettering requires quorum queues.");
+                    if (_options.Overflow != QueueOverflowBehavior.RejectPublish)
+                        throw new MessageBusException("At-least-once broker dead-lettering requires RejectPublish overflow.");
+                }
+                arguments[RabbitMQConstants.DeadLetterStrategyArgument] = _options.DeadLetterStrategy.Value.ToEnumString();
+            }
+        }
+        if (_options.Overflow.HasValue)
+            arguments[RabbitMQConstants.OverflowArgument] = _options.Overflow.Value.ToEnumString();
+        if (_options.ConsumerTimeout.HasValue)
+        {
+            if (!_isQuorumQueue)
+                throw new MessageBusException("Per-queue consumer timeout (x-consumer-timeout) requires quorum queues (RabbitMQ 4.3+). Call UseQuorumQueues() before ConsumerTimeout().");
+            if (serverVersion is not null && serverVersion < _delayedExchangePluginIncompatibleVersion)
+                throw new MessageBusException(FormattableString.Invariant($"Per-queue consumer timeout (x-consumer-timeout) requires RabbitMQ 4.3+. Detected server version: {serverVersion}."));
+            arguments[RabbitMQConstants.ConsumerTimeoutArgument] = (long)_options.ConsumerTimeout.Value.TotalMilliseconds;
+        }
+        if (_options.SingleActiveConsumer)
+            arguments[RabbitMQConstants.SingleActiveConsumerArgument] = true;
+        if (_options.MaxPriority.HasValue)
+            arguments[RabbitMQConstants.MaxPriorityArgument] = (int)_options.MaxPriority.Value;
+        if (_options.DelayedRetryType.HasValue)
+        {
+            if (!_isQuorumQueue)
+                throw new MessageBusException("Delayed retries (x-delayed-retry-*) require quorum queues (RabbitMQ 4.3+). Call UseQuorumQueues() before UseDelayedRetries().");
+            if (serverVersion is not null && serverVersion < _delayedExchangePluginIncompatibleVersion)
+                throw new MessageBusException(FormattableString.Invariant($"Delayed retries (x-delayed-retry-*) require RabbitMQ 4.3+. Detected server version: {serverVersion}."));
+            arguments[RabbitMQConstants.DelayedRetryTypeArgument] = _options.DelayedRetryType.Value.ToEnumString();
+            if (_options.DelayedRetryMin.HasValue)
+                arguments[RabbitMQConstants.DelayedRetryMinArgument] = _options.DelayedRetryMin.Value;
+            if (_options.DelayedRetryMax.HasValue)
+                arguments[RabbitMQConstants.DelayedRetryMaxArgument] = _options.DelayedRetryMax.Value;
+        }
+        var queue = await channel.QueueDeclareAsync(_options.SubscriptionQueueName, _options.IsDurable,
+            _options.IsSubscriptionQueueExclusive, _options.SubscriptionQueueAutoDelete,
+            arguments.Count > 0 ? arguments : null, cancellationToken: cancellationToken).AnyContext();
+        await channel.QueueBindAsync(queue.QueueName, _options.Topic, String.Empty, cancellationToken: cancellationToken).AnyContext();
+        return queue.QueueName;
+    }
+
+    internal async Task ConfigurePrefetchAsync(IChannel channel, Version? serverVersion, CancellationToken cancellationToken)
+    {
+        if (_options.PrefetchCount > 0 || _options.PrefetchSize > 0)
+        {
+#pragma warning disable CS0618
+            bool global = _options.GlobalQos;
+#pragma warning restore CS0618
+            if (global && (_isQuorumQueue || (serverVersion is not null && serverVersion >= _globalQosRemovedVersion)))
+            {
+                _logger.LogWarning("Global QoS is unavailable; using per-consumer prefetch");
+                global = false;
+            }
+            await channel.BasicQosAsync(_options.PrefetchSize, _options.PrefetchCount, global, cancellationToken).AnyContext();
+        }
+    }
+}

src/Foundatio.RabbitMQ/Utility/RabbitMQConstants.cs

@@ -0,0 +1,74 @@
+namespace Foundatio.Utility;
+
+/// <summary>RabbitMQ wire header and queue argument names shared by publishers, consumers, and topology setup.</summary>
+public static class RabbitMQConstants
+{
+    /// <summary>The <c>x-delivery-count</c> wire name.</summary>
+    public const string XDeliveryCountHeader = "x-delivery-count";
+
+    /// <summary>The <c>x-original-message-id</c> wire name.</summary>
+    public const string XOriginalMessageIdHeader = "x-original-message-id";
+
+    /// <summary>The <c>x-delay</c> wire name.</summary>
+    public const string XDelayHeader = "x-delay";
+
+    /// <summary>The <c>x-original-expiration</c> wire name.</summary>
+    public const string XOriginalExpirationHeader = "x-original-expiration";
+
+    /// <summary>The <c>x-foundatio-failure-type</c> wire name.</summary>
+    public const string FailureTypeHeader = "x-foundatio-failure-type";
+
+    /// <summary>The <c>x-foundatio-original-exchange</c> wire name.</summary>
+    public const string OriginalExchangeHeader = "x-foundatio-original-exchange";
+
+    /// <summary>The <c>x-foundatio-original-routing-key</c> wire name.</summary>
+    public const string OriginalRoutingKeyHeader = "x-foundatio-original-routing-key";
+
+    /// <summary>The <c>CC</c> wire name.</summary>
+    public const string CarbonCopyHeader = "CC";
+
+    /// <summary>The <c>BCC</c> wire name.</summary>
+    public const string BlindCarbonCopyHeader = "BCC";
+
+    /// <summary>The <c>x-queue-type</c> wire name.</summary>
+    public const string QueueTypeArgument = "x-queue-type";
+
+    /// <summary>The <c>x-delivery-limit</c> wire name.</summary>
+    public const string DeliveryLimitArgument = "x-delivery-limit";
+
+    /// <summary>The <c>x-dead-letter-exchange</c> wire name.</summary>
+    public const string DeadLetterExchangeArgument = "x-dead-letter-exchange";
+
+    /// <summary>The <c>x-dead-letter-routing-key</c> wire name.</summary>
+    public const string DeadLetterRoutingKeyArgument = "x-dead-letter-routing-key";
+
+    /// <summary>The <c>x-dead-letter-strategy</c> wire name.</summary>
+    public const string DeadLetterStrategyArgument = "x-dead-letter-strategy";
+
+    /// <summary>The <c>x-overflow</c> wire name.</summary>
+    public const string OverflowArgument = "x-overflow";
+
+    /// <summary>The <c>x-consumer-timeout</c> wire name.</summary>
+    public const string ConsumerTimeoutArgument = "x-consumer-timeout";
+
+    /// <summary>The <c>x-single-active-consumer</c> wire name.</summary>
+    public const string SingleActiveConsumerArgument = "x-single-active-consumer";
+
+    /// <summary>The <c>x-max-priority</c> wire name.</summary>
+    public const string MaxPriorityArgument = "x-max-priority";
+
+    /// <summary>The <c>x-delayed-retry-type</c> wire name.</summary>
+    public const string DelayedRetryTypeArgument = "x-delayed-retry-type";
+
+    /// <summary>The <c>x-delayed-retry-min</c> wire name.</summary>
+    public const string DelayedRetryMinArgument = "x-delayed-retry-min";
+
+    /// <summary>The <c>x-delayed-retry-max</c> wire name.</summary>
+    public const string DelayedRetryMaxArgument = "x-delayed-retry-max";
+
+    /// <summary>The <c>x-delayed-type</c> wire name.</summary>
+    public const string DelayedExchangeTypeArgument = "x-delayed-type";
+
+    /// <summary>The <c>x-max-length-bytes</c> wire name.</summary>
+    public const string MaxLengthBytesArgument = "x-max-length-bytes";
+}

tests/Foundatio.RabbitMQ.ProcessProbe/Directory.Build.props

@@ -0,0 +1,9 @@
+<Project>
+  <PropertyGroup>
+    <TargetFramework>net10.0</TargetFramework>
+    <LangVersion>latest</LangVersion>
+    <Nullable>enable</Nullable>
+    <TreatWarningsAsErrors>true</TreatWarningsAsErrors>
+    <IsPackable>false</IsPackable>
+  </PropertyGroup>
+</Project>

tests/Foundatio.RabbitMQ.ProcessProbe/Foundatio.RabbitMQ.ProcessProbe.csproj

@@ -0,0 +1,8 @@
+<Project Sdk="Microsoft.NET.Sdk">
+  <PropertyGroup>
+    <OutputType>Exe</OutputType>
+  </PropertyGroup>
+  <ItemGroup>
+    <ProjectReference Include="..\..\src\Foundatio.RabbitMQ\Foundatio.RabbitMQ.csproj" />
+  </ItemGroup>
+</Project>