Publish MaxRetries error-queue messages as persistent with the original properties

CHANGELOG.md

@@ -1,5 +1,15 @@
 # Changelog
 
+## 0.28.0 (2026-10-28)
+
+### Fixed
+
+- `Ears::Middlewares::MaxRetries` now publishes the message to the error queue as persistent. Previously the message was transient and a broker restart dropped it from a classic queue.
+
+### Changed
+
+- `Ears::Middlewares::MaxRetries` now copies the properties of the original message (headers, `content_type`, `message_id`, `timestamp` and so on) to the error-queue message. It does not copy `user_id` and removes the `x-death` header, so a message that you move back to the main queue is processed again.
+
 ## 0.27.0 (2026-10-02)
 
 ### Changed

Gemfile.lock

@@ -1,7 +1,7 @@
 PATH
   remote: .
   specs:
-    ears (0.27.0)
+    ears (0.28.0)
       bunny (>= 3.0.0)
       connection_pool (~> 3.0)
       json (>= 2.9.0)
@@ -102,7 +102,7 @@ CHECKSUMS
   bunny (3.3.0) sha256=85ef5562ec2b24f542e8fad69c4302d30d966fb79ff803dc9d0c398a4054672a
   connection_pool (3.0.2) sha256=33fff5ba71a12d2aa26cb72b1db8bba2a1a01823559fb01d29eb74c286e62e0a
   diff-lcs (1.6.2) sha256=9ae0d2cba7d4df3075fe8cd8602a8604993efc0dfa934cff568969efb1909962
-  ears (0.27.0)
+  ears (0.28.0)
   json (2.21.2) sha256=1f1d3b7cf2b3ba1a69beca0bb6db13d5438b80bff3cd54cdaaa620b9b07c1c6a
   language_server-protocol (3.17.0.6) sha256=5ef2c0c138f8267e1bc631d3328347d354f96724b0af22f2c79516120443b7f0
   lint_roller (1.1.0) sha256=2c0c845b632a7d172cb849cc90c1bce937a28c5c8ccccb50dfd46a485003cc87

README.md

@@ -509,6 +509,8 @@ end
 
 This will automatically route messages to `my_queue.error` after they have been re-tried three times. This prevents you from infinitely retrying a faulty message.
 
+The message in the error queue is persistent and keeps the properties and headers of the original message, except `user_id` and the `x-death` header. Register `MaxRetries` before any middleware that changes the body (for example `JSON`), because otherwise `MaxRetries` gets the parsed body, not the original text.
+
 ### Stopping any running consumers
 
 When you are running Ears in a non-blocking way (e.g. in a Thread), it might be cumbersome to remove the running consumers without restarting the whole app.

lib/ears/middlewares/max_retries.rb

@@ -4,6 +4,8 @@ module Ears
   module Middlewares
     # A middleware that automatically puts messages on an error queue when the specified number of retries are exceeded.
     class MaxRetries < Middleware
+      PERSISTENT_DELIVERY_MODE = 2
+
       # @param [Hash] opts The options for the middleware.
       # @option opts [Integer] :retries The number of retries before the message is sent to the error queue.
       # @option opts [String] :error_queue The name of the queue where messages should be sent to when the max retries are reached.
@@ -14,7 +16,7 @@ def initialize(opts)
       end
 
       def call(delivery_info, metadata, payload, app)
-        return handle_exceeded(payload) if retries_exceeded?(metadata)
+        return handle_exceeded(metadata, payload) if retries_exceeded?(metadata)
 
         app.call(delivery_info, metadata, payload)
       end
@@ -23,14 +25,27 @@ def call(delivery_info, metadata, payload, app)
 
       attr_reader :retries, :error_queue
 
-      def handle_exceeded(payload)
+      def handle_exceeded(metadata, payload)
         Bunny::Exchange.default(Ears.channel).publish(
           payload,
-          routing_key: error_queue,
+          **error_queue_options(metadata),
         )
         :ack
       end
 
+      def error_queue_options(metadata)
+        metadata
+          .to_hash
+          .except(:user_id)
+          .merge(
+            headers: metadata.headers.except('x-death'),
+            delivery_mode: PERSISTENT_DELIVERY_MODE,
+            timestamp: metadata.timestamp&.to_i,
+            routing_key: error_queue,
+          )
+          .compact
+      end
+
       def retries_exceeded?(metadata)
         return false if metadata.headers.nil?
 

lib/ears/version.rb

@@ -1,3 +1,3 @@
 module Ears
-  VERSION = '0.27.0'
+  VERSION = '0.28.0'
 end

spec/ears/middlewares/max_retries_spec.rb

@@ -4,9 +4,8 @@
 
 RSpec.describe Ears::Middlewares::MaxRetries do
   let(:delivery_info) { instance_double(Bunny::DeliveryInfo) }
-  let(:metadata) do
-    instance_double(
-      Bunny::MessageProperties,
+  let(:properties) do
+    {
       headers: {
         'x-death' => [
           {
@@ -26,9 +25,24 @@
             'routing-keys' => [''],
           },
         ],
+        'x-first-death-queue' => 'my_queue',
+        'x-last-death-reason' => 'rejected',
+        'tenant' => 'acme',
       },
-    )
+      delivery_mode: 2,
+      content_type: 'application/json',
+      content_encoding: 'utf-8',
+      priority: 5,
+      correlation_id: 'correlation-1',
+      reply_to: 'reply-queue',
+      message_id: 'message-1',
+      timestamp: Time.parse('2021-09-10T10:56:00Z'),
+      type: 'order.created',
+      user_id: 'original-user',
+      app_id: 'order-service',
+    }
   end
+  let(:metadata) { Bunny::MessageProperties.new(properties) }
   let(:payload) { 'payload' }
   let(:error_queue) { 'my_queue.error' }
   let(:middleware) do
@@ -56,30 +70,155 @@
 
   context 'when retry count is exceeded' do
     let(:max_retries) { 1 }
+    let(:published_options) { [] }
 
-    it 'acks the message when the max retry count is exceeded' do
-      allow(default_exchange).to receive(:publish)
+    before do
+      allow(default_exchange).to receive(:publish) do |_payload, **options|
+        published_options << options
+      end
+    end
 
+    it 'acks the message when the max retry count is exceeded' do
       expect(
         middleware.call(delivery_info, metadata, payload, Proc.new { :moep }),
       ).to eq(:ack)
     end
 
-    it 'publishes the message to the configured error exchange' do
+    it 'publishes the unchanged payload to the configured error queue' do
       expect(Bunny::Exchange).to receive(:default).with(channel).and_return(
         default_exchange,
       )
-      expect(default_exchange).to receive(:publish).with(
+
+      middleware.call(delivery_info, metadata, payload, Proc.new { :moep })
+
+      expect(default_exchange).to have_received(:publish).with(
         payload,
-        routing_key: error_queue,
+        hash_including(routing_key: error_queue),
       )
+    end
 
+    it 'publishes the message as persistent' do
       middleware.call(delivery_info, metadata, payload, Proc.new { :moep })
+
+      expect(published_options.first).to include(delivery_mode: 2)
+    end
+
+    it 'publishes as persistent when the original message was transient' do
+      properties[:delivery_mode] = 1
+
+      middleware.call(delivery_info, metadata, payload, Proc.new { :moep })
+
+      expect(published_options.first).to include(delivery_mode: 2)
+    end
+
+    it 'copies the properties of the original message' do
+      middleware.call(delivery_info, metadata, payload, Proc.new { :moep })
+
+      expect(published_options.first).to include(
+        content_type: 'application/json',
+        content_encoding: 'utf-8',
+        priority: 5,
+        correlation_id: 'correlation-1',
+        reply_to: 'reply-queue',
+        message_id: 'message-1',
+        timestamp: Time.parse('2021-09-10T10:56:00Z').to_i,
+        type: 'order.created',
+        app_id: 'order-service',
+      )
+    end
+
+    it 'removes the x-death header and keeps the other headers' do
+      middleware.call(delivery_info, metadata, payload, Proc.new { :moep })
+
+      expect(published_options.first[:headers]).to eq(
+        'x-first-death-queue' => 'my_queue',
+        'x-last-death-reason' => 'rejected',
+        'tenant' => 'acme',
+      )
+    end
+
+    it 'does not copy the user id' do
+      middleware.call(delivery_info, metadata, payload, Proc.new { :moep })
+
+      expect(published_options.first).not_to have_key(:user_id)
+    end
+
+    it 'does not change the metadata of the original message' do
+      middleware.call(delivery_info, metadata, payload, Proc.new { :moep })
+
+      expect(properties[:headers]).to have_key('x-death')
+      expect(properties).to include(user_id: 'original-user', delivery_mode: 2)
+    end
+  end
+
+  context 'when publishing through a real Bunny channel' do
+    let(:max_retries) { 1 }
+    let(:connection) do
+      instance_double(
+        Bunny::Session,
+        logger: Logger.new(nil),
+        next_channel_id: 1,
+        register_channel: nil,
+        mutex_impl: Mutex,
+        transport_write_timeout: 1,
+        send_frameset: nil,
+        record_exchange: nil,
+      )
+    end
+    let(:channel) { Bunny::Channel.new(connection) }
+    let(:default_exchange) { Bunny::Exchange.default(channel) }
+    let(:header_frame) do
+      sent_frames = nil
+      allow(connection).to receive(:send_frameset) do |frames, _channel|
+        sent_frames = frames
+      end
+
+      middleware.call(delivery_info, metadata, payload, Proc.new { :moep })
+
+      sent_frames.find { |frame| frame.is_a?(AMQ::Protocol::HeaderFrame) }
+    end
+
+    before do
+      channel.instance_variable_set(:@status, :open)
+      channel.instance_variable_set(:@frame_max, 131_072)
+      allow(Bunny::Exchange).to receive(:default).and_call_original
+    end
+
+    it 'sends the message with delivery mode persistent' do
+      properties[:delivery_mode] = 1
+
+      expect(header_frame.properties[:delivery_mode]).to eq(2)
+    end
+
+    it 'sends the copied properties and headers without x-death' do
+      expect(header_frame.properties).to include(
+        message_id: 'message-1',
+        content_type: 'application/json',
+        headers: hash_excluding('x-death'),
+      )
+    end
+
+    it 'sends the original timestamp' do
+      expect(header_frame.properties[:timestamp]).to eq(
+        Time.parse('2021-09-10T10:56:00Z'),
+      )
+    end
+
+    it 'sends no user id' do
+      expect(header_frame.properties).not_to have_key(:user_id)
+    end
+
+    context 'when the original message has no timestamp' do
+      before { properties.delete(:timestamp) }
+
+      it 'sends the message without a timestamp' do
+        expect(header_frame.properties).not_to have_key(:timestamp)
+      end
     end
   end
 
   context 'when no death headers are present' do
-    let(:metadata) { instance_double(Bunny::MessageProperties, headers: {}) }
+    let(:metadata) { Bunny::MessageProperties.new({ headers: {} }) }
 
     it 'returns downstream result if no death headers are present' do
       expect(
@@ -89,7 +228,7 @@
   end
 
   context 'when metadata has no headers' do
-    let(:metadata) { instance_double(Bunny::MessageProperties, headers: nil) }
+    let(:metadata) { Bunny::MessageProperties.new({ headers: nil }) }
 
     it 'returns downstream result if no death headers are present' do
       expect(