don't let Launcher know about clustering/leases

Let it be a concern for Clustering Controller

No need to poll the data dir lock, because it's only required for NFS
disks.

src/lavinmq.cr

@@ -17,12 +17,8 @@ config.parse # both ARGV and config file
 require "./lavinmq/launcher"
 require "./lavinmq/clustering/controller"
 
-begin
-  if config.clustering?
-    LavinMQ::Clustering::Controller.new(config).run
-  else
-    LavinMQ::Launcher.new(config).run
-  end
-rescue LavinMQ::Launcher::LostLeadership
-  exit 3
+if config.clustering?
+  LavinMQ::Clustering::Controller.new(config).run
+else
+  LavinMQ::Launcher.new(config).run
 end

src/lavinmq/clustering/controller.cr

@@ -17,15 +17,27 @@ class LavinMQ::Clustering::Controller
   def run
     spawn(follow_leader, name: "Follower monitor")
     wait_to_be_insync
-    lease = @etcd.elect("#{@config.clustering_etcd_prefix}/leader", @advertised_uri) # blocks until becoming leader
+    @lease = lease = @etcd.elect("#{@config.clustering_etcd_prefix}/leader", @advertised_uri) # blocks until becoming leader
     # TODO: make sure we still are in the ISR set
     replicator = Clustering::Server.new(@config, @etcd)
-    @launcher = l = Launcher.new(@config, replicator, lease)
-    l.run
+    @launcher = Launcher.new(@config, replicator).start
+    loop do
+      if lease.wait(30.seconds)
+        break if @stopped
+        Log.fatal { "Lost cluster leadership" }
+        exit 3
+      else
+        GC.collect
+      end
+    end
   end
 
+  @stopped = false
+
   def stop
+    @stopped = true
     @launcher.try &.stop
+    @lease.try &.release
   end
 
   # Each node in a cluster has an unique id, for tracking ISR

src/lavinmq/launcher.cr

@@ -6,7 +6,6 @@ require "./server"
 require "./http/http_server"
 require "./in_memory_backend"
 require "./data_dir_lock"
-require "./etcd"
 
 module LavinMQ
   class Launcher
@@ -15,9 +14,8 @@ module LavinMQ
     @first_shutdown_attempt = true
     @data_dir_lock : DataDirLock?
     @closed = false
-    @leadership : Etcd::Leadership?
 
-    def initialize(@config : Config, replicator = Clustering::NoopServer.new, @leadership = nil)
+    def initialize(@config : Config, replicator = Clustering::NoopServer.new)
       print_environment_info
       print_max_map_count
       fd_limit = System.maximize_fd_limit
@@ -38,23 +36,17 @@ module LavinMQ
       setup_log_exchange
     end
 
-    def run
+    def start : self
       listen
       SystemD.notify_ready
+      self
+    end
+
+    def run
+      start
       loop do
-        if leadership = @leadership
-          if leadership.wait(30.seconds)
-            Log.fatal { "Lost cluster leadership" }
-            raise LostLeadership.new
-          else
-            @data_dir_lock.try &.poll
-            GC.collect
-          end
-        else
-          sleep 30.seconds
-          @data_dir_lock.try &.poll
-          GC.collect
-        end
+        sleep 30.seconds
+        GC.collect
       end
     end
 
@@ -66,7 +58,6 @@ module LavinMQ
       @http_server.close rescue nil
       @amqp_server.close rescue nil
       @data_dir_lock.try &.release
-      @leadership.try &.release
     end
 
     private def print_environment_info