From e46e2fb1a1584f377159621340948d560f1351b9 Mon Sep 17 00:00:00 2001 From: AlexMarold Date: Wed, 28 Oct 2026 11:57:53 +0100 Subject: [PATCH] Publish error-queue messages as persistent and keep the original properties A broker restart dropped messages that MaxRetries had moved to the error queue on classic queues, because it published only the payload as a transient message. It also lost the headers, content type, message id and timestamp of the original message. MaxRetries now publishes with delivery mode 2 and copies the properties of the received message. It removes the x-death header, so a message that is moved back to the main queue is processed again, and it does not copy user_id, because RabbitMQ rejects a publish whose user_id differs from the connection user. The timestamp is converted from Time to an integer, because Bunny decodes it as Time but cannot encode it again. The delivery mode is set after the copy, because Bunny only derives it when the options do not contain one. Messages that are already in an error queue stay transient. MaxRetries still acks without publisher confirms. The release is 0.28.0. Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 10 ++ Gemfile.lock | 4 +- README.md | 2 + lib/ears/middlewares/max_retries.rb | 21 ++- lib/ears/version.rb | 2 +- spec/ears/middlewares/max_retries_spec.rb | 161 ++++++++++++++++++++-- 6 files changed, 183 insertions(+), 17 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 078021d..65fffb3 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/Gemfile.lock b/Gemfile.lock index 27e861d..f35f0fe 100644 --- a/Gemfile.lock +++ b/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 diff --git a/README.md b/README.md index 9ce327e..f6edb6a 100644 --- a/README.md +++ b/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. diff --git a/lib/ears/middlewares/max_retries.rb b/lib/ears/middlewares/max_retries.rb index 8490056..d207fbc 100644 --- a/lib/ears/middlewares/max_retries.rb +++ b/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? diff --git a/lib/ears/version.rb b/lib/ears/version.rb index 73f5a30..39ba831 100644 --- a/lib/ears/version.rb +++ b/lib/ears/version.rb @@ -1,3 +1,3 @@ module Ears - VERSION = '0.27.0' + VERSION = '0.28.0' end diff --git a/spec/ears/middlewares/max_retries_spec.rb b/spec/ears/middlewares/max_retries_spec.rb index 2891840..31d963f 100644 --- a/spec/ears/middlewares/max_retries_spec.rb +++ b/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(