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(