Bugfix: dont expire messages before server is fully started (#1714)

### WHAT is this pull request doing?
Make message expire fibers wait for vhost to be fully loaded before expiring messages.

Fixes #1697 

### HOW can this pull request be tested?
Specs, but manually never hurts

Author: spuun

spec/queue_spec.cr

@@ -16,6 +16,34 @@ def with_queue(&)
 end
 
 describe LavinMQ::AMQP::Queue do
+  it "should not expire message before server is fully started" do
+    # https://github.com/cloudamqp/lavinmq/issues/1697
+    with_amqp_server do |s|
+      with_channel(s) do |ch|
+        # Must go via a queue with binding keys
+        dlq = ch.queue("dlq")
+        dlq.bind(ch.topic_exchange.name, dlq.name)
+        q = ch.queue("ttl", args: AMQP::Client::Arguments.new(
+          {"x-message-ttl" => 1000, "x-dead-letter-exchange" => "amq.topic", "x-dead-letter-routing-key" => dlq.name}
+        ))
+        # declare a queue after ttl that will cause yields in its message store during load
+        other_queue = ch.queue("other")
+        x = ch.default_exchange
+        x.publish_confirm("foo", other_queue.name)
+        x.publish_confirm("ttl", q.name)
+      end
+      s.stop
+      RoughTime.paused do |t|
+        # Move time so message will be expired on startup
+        t.travel 2.seconds
+        s.restart
+        with_channel(s) do |ch|
+          ch.queue("dlq").get.should_not be_nil, failure_message: "Message not dead lettered?!"
+        end
+      end
+    end
+  end
+
   it "should expire itself after last consumer disconnects" do
     with_amqp_server do |s|
       with_channel(s) do |ch|

src/lavinmq/amqp/queue/queue.cr

@@ -92,6 +92,7 @@ module LavinMQ::AMQP
     getter? internal = false
 
     private def queue_expire_loop
+      @vhost.closed.when_false.receive?
       loop do
         break unless @expires
         @consumers_empty.when_true.receive
@@ -110,6 +111,7 @@ module LavinMQ::AMQP
     end
 
     private def message_expire_loop
+      @vhost.closed.when_false.receive?
       loop do
         @consumers_empty.when_true.receive
         @log.debug { "Consumers empty" }

src/lavinmq/bool_channel.cr

@@ -48,11 +48,22 @@ class BoolChannel
   end
 
   def value
-    @value.get
+    @value.get(:acquire)
   end
 
-  def set(value : Bool)
-    return if @value.swap(value) == value
+  def swap(value : Bool)
+    ret = @value.swap(value)
+    update_active_channel(value) unless ret == value
+    ret
+  end
+
+  def set(value : Bool) : Nil
+    @value.set(value)
+    update_active_channel(value)
+    nil
+  end
+
+  private def update_active_channel(value)
     if value
       @when_false.deactivate
       @when_true.activate

src/lavinmq/server.cr

@@ -29,11 +29,11 @@ module LavinMQ
     end
 
     getter vhosts, users, data_dir, parameters, authenticator
-    getter? closed, flow
+    getter? flow
     include ParameterTarget
 
     @start = Time.instant
-    @closed = false
+    @closed = BoolChannel.new(false)
     @flow = true
     @listeners = Hash(Socket::Server, Protocol).new # Socket => protocol
     @connection_factories = Hash(Protocol, ConnectionFactory).new
@@ -46,6 +46,7 @@ module LavinMQ
       Schema.migrate(@data_dir, @replicator)
       @users = Auth::UserStore.new(@data_dir, @replicator)
       @vhosts = VHostStore.new(@data_dir, @users, @replicator)
+      @vhosts.load!
       @mqtt_brokers = MQTT::Brokers.new(@vhosts, @replicator)
       @parameters = ParameterStore(Parameter).new(@data_dir, "parameters.json", @replicator)
       @authenticator = Auth::Chain.create(@config, @users)
@@ -57,6 +58,10 @@ module LavinMQ
       spawn stats_loop, name: "Server#stats_loop"
     end
 
+    def closed?
+      @closed.value
+    end
+
     def followers
       @replicator.try(&.followers) || Array(Clustering::Follower).new
     end
@@ -80,8 +85,7 @@ module LavinMQ
     end
 
     def stop
-      return if @closed
-      @closed = true
+      return if @closed.swap(true)
       @vhosts.close
       @replicator.try &.clear
       @authenticator.try &.cleanup
@@ -95,11 +99,12 @@ module LavinMQ
       @users = Auth::UserStore.new(@data_dir, @replicator)
       @authenticator = Auth::Chain.create(@config, @users)
       @vhosts = VHostStore.new(@data_dir, @users, @replicator)
+      @vhosts.load!
       @connection_factories[Protocol::AMQP] = AMQP::ConnectionFactory.new(@authenticator, @vhosts)
       @connection_factories[Protocol::MQTT] = MQTT::ConnectionFactory.new(@authenticator, @mqtt_brokers, @config)
       @parameters = ParameterStore(Parameter).new(@data_dir, "parameters.json", @replicator)
       apply_parameter
-      @closed = false
+      @closed.set(false)
       Fiber.yield
     end
 
@@ -112,7 +117,7 @@ module LavinMQ
       Log.info { "Listening for #{protocol} on #{s.local_address}" }
       loop do
         client = s.accept? || break
-        next client.close if @closed
+        next client.close if closed?
         accept_tcp(client, protocol)
       end
     rescue ex : IO::Error
@@ -162,7 +167,7 @@ module LavinMQ
       Log.info { "Listening for #{protocol} on #{s.local_address}" }
       loop do # do not try to use while
         client = s.accept? || break
-        next client.close if @closed
+        next client.close if closed?
         accept_unix(client, protocol)
       end
     rescue ex : IO::Error
@@ -198,7 +203,7 @@ module LavinMQ
       Log.info { "Listening for #{protocol} on #{s.local_address} (TLS)" }
       loop do # do not try to use while
         client = s.accept? || break
-        next client.close if @closed
+        next client.close if closed?
         accept_tls(client, context, protocol)
       end
     rescue ex : IO::Error | OpenSSL::Error
@@ -255,7 +260,7 @@ module LavinMQ
     end
 
     def close
-      @closed = true
+      @closed.set(true)
       Log.debug { "Closing listeners" }
       @listeners.each_key &.close
       Log.debug { "Closing vhosts" }

src/lavinmq/vhost.cr

@@ -28,7 +28,12 @@ module LavinMQ
     getter name, exchanges, queues, data_dir, operator_policies, policies, parameters, shovels,
       direct_reply_consumers, connections, dir, users
     property? flow = true
-    getter? closed = false
+    getter closed = BoolChannel.new(true)
+
+    def closed?
+      @closed.value
+    end
+
     property max_connections : Int32?
     property max_queues : Int32?
 
@@ -66,8 +71,11 @@ module LavinMQ
 
     private def check_consumer_timeouts_loop
       loop do
-        sleep Config.instance.consumer_timeout_loop_interval.seconds
-        return if @closed
+        select
+        when timeout Config.instance.consumer_timeout_loop_interval.seconds
+        when closed.when_true.receive?
+          return
+        end
         @connections.each do |c|
           c.channels.each_value do |ch|
             ch.check_consumer_timeout
@@ -401,7 +409,7 @@ module LavinMQ
     end
 
     def close(reason = "Broker shutdown")
-      @closed = true
+      return if @closed.swap(true)
       stop_shovels
       stop_upstream_links
 
@@ -475,11 +483,12 @@ module LavinMQ
       if has_parameters || has_policies
         spawn(name: "Load parameters") do
           sleep 10.milliseconds
-          next if @closed
+          next if closed?
           apply_parameters
           apply_policies
         end
       end
+      closed.set(false)
       Fiber.yield
     end
 

src/lavinmq/vhost_store.cr

@@ -17,7 +17,6 @@ module LavinMQ
 
     def initialize(@data_dir : String, @users : Auth::UserStore, @replicator : Clustering::Replicator?)
       @vhosts = Hash(String, VHost).new
-      load!
     end
 
     forward_missing_to @vhosts
@@ -72,7 +71,7 @@ module LavinMQ
       end
     end
 
-    private def load!
+    def load!
       path = File.join(@data_dir, "vhosts.json")
       if File.exists? path
         Log.debug { "Loading vhosts from file" }
@@ -90,6 +89,8 @@ module LavinMQ
         Log.debug { "Loading default vhosts" }
         create("/")
       end
+      # Wait for vhosts to be loaded ("not closed")
+      @vhosts.each_value &.closed.when_false.receive
       Log.debug { "#{size} vhosts loaded" }
     rescue ex
       Log.error(exception: ex) { "Failed to load vhosts" }