Always use x-dead-letter-exchange when deliver a delayed message

CHANGELOG.md

@@ -15,6 +15,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - Wait for followers to synchronize on shutdown of leader
 - Shovel the exact number of messages available on start if `delete-after=queue-length`, not more
 - Prevet a queue that's overflowing to consume too much resources
+- Dead-lettering loop when publishing to a delayed exchange's internal queue [#748](https://github.com/cloudamqp/lavinmq/pull/748)
 
 ### Changed
 

spec/delayed_message_exchange_spec.cr

@@ -119,4 +119,39 @@ describe "Delayed Message Exchange" do
       end
     end
   end
+
+  it "should always use x-dead-letter-exchange when expiring the message" do
+    with_amqp_server do |s|
+      with_channel(s) do |ch|
+        x = ch.exchange(x_name, "topic", args: x_args)
+        q = ch.queue(q_name)
+        q.bind(x.name, "#")
+
+        delayed_q_name = s.vhosts["/"].exchanges[x_name].@delayed_queue.try &.name || raise "No delay queue?"
+
+        msgs = Channel(AMQP::Client::DeliverMessage).new
+        q.subscribe { |msg| msgs.send msg }
+
+        props = AMQP::Client::Properties.new(
+          headers: AMQP::Client::Arguments.new({
+            "x-delay" => 100,
+          }),
+        )
+
+        # Publish direct to the delayed queue which leaves exchange empty
+        ch.basic_publish "body", exchange: "", routing_key: delayed_q_name, props: props
+
+        # Wait for the message to be expired. Just receiving it actually verifies that
+        # exchange has been set to x-dead-letter-exchange
+        select
+        when msg = msgs.receive
+          msg.exchange.should eq x_name
+        when timeout 3.second
+          msgs.close
+          ch.close
+          raise "x-dead-letter-exchange not sent, message is looping?"
+        end
+      end
+    end
+  end
 end

src/lavinmq/queue/delayed_exchange_queue.cr

@@ -5,6 +5,13 @@ module LavinMQ
   class DelayedExchangeQueue < Queue
     @internal = true
 
+    @exchange_name : String
+
+    def initialize(*args)
+      super(*args)
+      @exchange_name = arguments["x-dead-letter-exchange"]?.try(&.to_s) || raise "Missing x-dead-letter-exchange"
+    end
+
     private def init_msg_store(data_dir)
       replicator = durable? ? @vhost.@replicator : nil
       DelayedMessageStore.new(data_dir, replicator)
@@ -44,7 +51,7 @@ module LavinMQ
         headers.delete("x-delay")
         msg.properties.headers = headers
       end
-      @vhost.publish Message.new(msg.timestamp, msg.exchange_name, msg.routing_key,
+      @vhost.publish Message.new(msg.timestamp, @exchange_name, msg.routing_key,
         msg.properties, msg.bodysize, IO::Memory.new(msg.body))
       delete_message sp
     end