Bugfix: Don't reset x-cache-ttl when expiring msg with delayed exchange

Previously, when using deduplication and delayed exchange together, the x-cache-ttl would be reset when a message was expired and re-published (if cache-ttl and delay was the same), meaning the cache_ttl was effectively more like cache_ttl + msg_delay. This changes the behavior so that a delayed message will not trigger deduplication by calling exchange.route_msg instead of vhost.publish.

Also refactors exchange.publish to separate the msg routing to a separate route_msg function (was do_publish) that just does the msg routing.

Author: viktorerlingsson

spec/deduplication_spec.cr

@@ -160,5 +160,44 @@ describe LavinMQ::Deduplication::Deduper do
       calls.first[0].should eq "msg1"
       calls.first[1].should eq 10
     end
+
+    it "should not reset x-cache-ttl on expire when using delayed exchange" do
+      cache_ttl = 1000
+      x_args = AMQP::Client::Arguments.new({
+        "x-delayed-exchange"      => true,
+        "x-delayed-type"          => "topic",
+        "x-message-deduplication" => true,
+        "x-cache-ttl"             => cache_ttl,
+        "x-cache-size"            => 1000,
+      })
+      q_name = "delayed_q"
+      with_amqp_server do |s|
+        with_channel(s) do |ch|
+          x = ch.exchange("delayed_ex", "topic", args: x_args)
+          q = ch.queue(q_name)
+          q.bind(x.name, "#")
+          hdrs = AMQP::Client::Arguments.new({
+            "x-delay"                => cache_ttl,
+            "x-deduplication-header" => "msg1",
+          })
+          x.publish "test message", "rk", props: AMQP::Client::Properties.new(headers: hdrs)
+          queue = s.vhosts["/"].queues[q_name]
+          queue.message_count.should eq 0 # no message yet, delayed exchange
+
+          # second publish should be deduplicated
+          sleep 100.milliseconds # wait for first publish to be processed
+          x.publish "test message", "rk", props: AMQP::Client::Properties.new(headers: hdrs)
+          wait_for { queue.message_count == 1 }
+
+          # get message, message_count should be 0
+          ch.basic_get(q_name, no_ack: true)
+          wait_for { queue.message_count == 0 }
+
+          # cache_ttl has passed, message should be delivered
+          x.publish "test message", "rk", props: AMQP::Client::Properties.new(headers: hdrs)
+          wait_for { queue.message_count == 1 }
+        end
+      end
+    end
   end
 end

spec/exchange_spec.cr

@@ -228,13 +228,13 @@ describe LavinMQ::Exchange do
             "x-deduplication-header" => "msg1",
           }))
           msg = LavinMQ::Message.new("ex", "rk", "body", props)
-          ex.publish(msg, false).should eq 1
+          ex.publish(msg, false).should eq true
           ex.dedup_count.should eq 0
           props = LavinMQ::AMQP::Properties.new(headers: LavinMQ::AMQP::Table.new({
             "x-deduplication-header" => "msg1",
           }))
           msg = LavinMQ::Message.new("ex", "rk", "body", props)
-          ex.publish(msg, false).should eq 0
+          ex.publish(msg, false).should eq false
           ex.dedup_count.should eq 1
 
           q.message_count.should eq 1
@@ -257,13 +257,13 @@ describe LavinMQ::Exchange do
             "custom" => "msg1",
           }))
           msg = LavinMQ::Message.new("ex", "rk", "body", props)
-          ex.publish(msg, false).should eq 1
+          ex.publish(msg, false).should eq true
           ex.dedup_count.should eq 0
           props = LavinMQ::AMQP::Properties.new(headers: LavinMQ::AMQP::Table.new({
             "custom" => "msg1",
           }))
           msg = LavinMQ::Message.new("ex", "rk", "body", props)
-          ex.publish(msg, false).should eq 0
+          ex.publish(msg, false).should eq false
           ex.dedup_count.should eq 1
 
           q.message_count.should eq 1

src/lavinmq/amqp/exchange/exchange.cr

@@ -196,43 +196,50 @@ module LavinMQ
 
       def publish(msg : Message, immediate : Bool,
                   queues : Set(LavinMQ::Queue) = Set(LavinMQ::Queue).new,
-                  exchanges : Set(LavinMQ::Exchange) = Set(LavinMQ::Exchange).new) : UInt32
+                  exchanges : Set(LavinMQ::Exchange) = Set(LavinMQ::Exchange).new) : Bool
         @publish_in_count.add(1, :relaxed)
         if d = @deduper
           if d.duplicate?(msg)
             @dedup_count.add(1, :relaxed)
-            return 0u32
+            return false
           end
           d.add(msg)
         end
-        count = do_publish(msg, immediate, queues, exchanges)
-        @unroutable_count.add(1, :relaxed) if count.zero?
-        @publish_out_count.add(count, :relaxed)
-        count
-      end
-
-      private def do_publish(msg : Message, immediate : Bool, queues : Set(LavinMQ::Queue), exchanges : Set(LavinMQ::Exchange)) : UInt32
-        headers = msg.properties.headers
-        if should_delay_message?(headers)
+        if should_delay_message?(msg.properties.headers)
           if q = @delayed_queue
             q.publish(msg)
-            return 1u32
+            @publish_out_count.add(1, :relaxed)
+            return true
           else
-            return 0u32
+            @unroutable_count.add(1, :relaxed)
+            return false
           end
         end
+        route_msg(msg, immediate, queues, exchanges)
+      end
+
+      def route_msg(msg : Message) : Bool
+        route_msg(msg, false, Set(LavinMQ::Queue).new, Set(LavinMQ::Exchange).new)
+      end
+
+      private def route_msg(msg : Message, immediate : Bool, queues : Set(LavinMQ::Queue), exchanges : Set(LavinMQ::Exchange)) : Bool
+        headers = msg.properties.headers
         find_queues(msg.routing_key, headers, queues, exchanges)
-        return 0u32 if queues.empty?
-        return 0u32 if immediate && !queues.any? &.immediate_delivery?
+        if queues.empty? || (immediate && !queues.any? &.immediate_delivery?)
+          @unroutable_count.add(1, :relaxed)
+          return false
+        end
 
         count = 0u32
         queues.each do |queue|
           if queue.publish(msg)
-            count += 1u32
+            count += 1
             msg.body_io.seek(-msg.bodysize.to_i64, IO::Seek::Current) # rewind
           end
         end
-        count
+        @publish_out_count.add(count, :relaxed)
+        @unroutable_count.add(1, :relaxed) if count.zero?
+        count.positive?
       end
 
       def find_queues(routing_key : String, headers : AMQP::Table?,

src/lavinmq/amqp/queue/delayed_exchange_queue.cr

@@ -49,7 +49,7 @@ module LavinMQ::AMQP
         headers.delete("x-delay")
         msg.properties.headers = headers
       end
-      @vhost.publish Message.new(msg.timestamp, @exchange_name, msg.routing_key,
+      @vhost.exchanges[@exchange_name].route_msg Message.new(msg.timestamp, @exchange_name, msg.routing_key,
         msg.properties, msg.bodysize, IO::Memory.new(msg.body))
       delete_message sp
     end

src/lavinmq/vhost.cr

@@ -124,8 +124,7 @@ module LavinMQ
     def publish(msg : Message, immediate = false,
                 visited = Set(LavinMQ::Exchange).new, found_queues = Set(LavinMQ::Queue).new) : Bool
       ex = @exchanges[msg.exchange_name]? || return false
-      published_queue_count = ex.publish(msg, immediate, found_queues, visited)
-      !published_queue_count.zero?
+      ex.publish(msg, immediate, found_queues, visited)
     ensure
       visited.clear
       found_queues.clear