Bugfix: dead letter cycle detection (#1723)

Fixes a bug in dead letter cycle detection where cycles were only
checked against the source queue rather than against each destination
queue individually. This could cause infinite loops when a dead letter
exchange routes to multiple queues.

The previous implementation checked for cycles once before routing to
any destination queues. It compared the source queue name against the
x-death header to detect cycles. However, when a dead letter exchange
routes to multiple queues, this approach was insufficient — each
destination queue needs to be checked.


Co-authored-by: Magnus Landerblom <[email protected]>

Author: spuun

spec/queue_dead_lettering_spec.cr

@@ -545,6 +545,74 @@ module DeadLetteringSpec
           msg.body_io.to_s.should eq "msg1"
         end
       end
+
+      it "should detect cycle with two queues in chain" do
+        with_amqp_server do |s|
+          with_channel(s) do |ch|
+            qargs1 = {
+              "x-max-length"              => 1,
+              "x-dead-letter-exchange"    => "",
+              "x-dead-letter-routing-key" => "q2",
+            }
+            q1 = ch.queue("q1", args: AMQP::Client::Arguments.new(qargs1))
+            qargs2 = {
+              "x-max-length"              => 1,
+              "x-dead-letter-exchange"    => "",
+              "x-dead-letter-routing-key" => "q1",
+            }
+            q2 = ch.queue("q2", args: AMQP::Client::Arguments.new(qargs2))
+            # "fill" queues
+            ch.default_exchange.publish("m1", "q1")
+            ch.default_exchange.publish("m2", "q2")
+            # now overflow q1
+            ch.default_exchange.publish("m3", "q1")
+            # m1 should be dead lettered to q2, which in turn is overflowed so
+            # m2 is dead letterd to q1. q1 overflowed again and m3 is dead lettered
+            # to q2. When q2 now overflows m1 shouldn't be dead lettered to q1,
+            # but instead be dropped.
+
+            get1(q1).body_io.to_s.should eq "m2"
+            get1(q2).body_io.to_s.should eq "m3"
+          end
+        end
+      end
+
+      it "should detect cycle with three queues in chain" do
+        with_amqp_server do |s|
+          with_channel(s) do |ch|
+            qargs1 = {
+              "x-max-length"              => 1,
+              "x-dead-letter-exchange"    => "",
+              "x-dead-letter-routing-key" => "q2",
+            }
+            q1 = ch.queue("q1", args: AMQP::Client::Arguments.new(qargs1))
+            qargs2 = {
+              "x-max-length"              => 1,
+              "x-dead-letter-exchange"    => "",
+              "x-dead-letter-routing-key" => "q3",
+            }
+            q2 = ch.queue("q2", args: AMQP::Client::Arguments.new(qargs2))
+            qargs3 = {
+              "x-max-length"              => 1,
+              "x-dead-letter-exchange"    => "",
+              "x-dead-letter-routing-key" => "q1",
+            }
+            q3 = ch.queue("q3", args: AMQP::Client::Arguments.new(qargs3))
+            # "fill" queues
+            ch.default_exchange.publish("m1", "q1")
+            ch.default_exchange.publish("m2", "q2")
+            ch.default_exchange.publish("m3", "q3")
+            # now overflow q1
+            ch.default_exchange.publish("m4", "q1")
+
+            # messages should end up in the queue before the queue they were
+            # published to
+            get1(q1).body_io.to_s.should eq "m2"
+            get1(q2).body_io.to_s.should eq "m3"
+            get1(q3).body_io.to_s.should eq "m4"
+          end
+        end
+      end
     end
 
     describe "Header Validation" do

src/lavinmq/amqp/argument/dead_lettering.cr

@@ -19,7 +19,7 @@ module LavinMQ::AMQP
         def initialize(@vhost : VHost, @queue_name : String, @log : Logger)
         end
 
-        private def cycle?(props, reason) : Bool
+        private def cycle?(qname, props, reason) : Bool
           unless headers = props.headers
             return false
           end
@@ -33,7 +33,7 @@ module LavinMQ::AMQP
           xdeaths.each do |field|
             next unless xdeath = field.as?(AMQ::Protocol::Table)
             return false if xdeath["reason"]? == "rejected"
-            return true if xdeath["queue"]? == @queue_name && xdeath["reason"] == reason.to_s
+            return true if xdeath["queue"]? == qname && xdeath["reason"] == reason.to_s
           end
           false
         end
@@ -42,7 +42,7 @@ module LavinMQ::AMQP
         # It's done like this to be able to dead letter to all destinations
         # except to the queue itself if a cycle is detected.
         # This is also how it's done in rabbitmq
-        def route(msg : BytesMessage, reason) # ameba:disable Metrics/CyclomaticComplexity
+        def route(msg : BytesMessage, reason)
           # No dead letter exchange => nothing to do
           return unless dlx = (msg.dlx || dlx())
           ex = @vhost.exchanges[dlx.to_s]?.as?(AMQP::Exchange) || return
@@ -71,14 +71,12 @@ module LavinMQ::AMQP
           ex.find_queues(routing_rk, routing_headers, queues)
           return if queues.empty?
 
-          is_cycle = cycle?(props, reason)
-
           dead_lettered_msg = Message.new(
             RoughTime.unix_ms, dlx.to_s, routing_rk.to_s,
             props, msg.bodysize, IO::Memory.new(msg.body))
 
           queues.each do |q|
-            next if is_cycle && q.name == @queue_name
+            next if cycle?(q.name, props, reason)
             @log.trace { "dead lettering dest=#{q.name} msg=#{dead_lettered_msg}" }
             q.publish(dead_lettered_msg)
           rescue ex