fix(mqtt): apply broker changes from the database seamlessly

spec/publishing/broker_changes_spec.cr

@@ -0,0 +1,166 @@
+require "../spec_helper"
+
+module PlaceOS::Source
+  # Brokers are added, edited and removed in the database while source runs.
+  # These go through the changefeed, as they do in production
+  describe MqttBrokerManager, tags: "broker-changes" do
+    managers = [] of MqttBrokerManager
+    brokers = [] of Model::Broker
+    relays = [] of BrokerRelay
+
+    # Every example has a Broker of its own, so the client ids never collide
+    new_broker = ->(port : Int32) {
+      shared = test_broker
+      broker = Model::Broker.new(
+        name: "changes-#{UUID.random}",
+        host: port == shared.port ? shared.host : "127.0.0.1",
+        port: port,
+        auth_type: :user_password,
+        username: "source",
+        password: "first",
+      ).save!
+      brokers << broker
+      broker
+    }
+
+    start_manager = -> {
+      manager = MqttBrokerManager.new
+      managers << manager
+      manager.start
+      manager
+    }
+
+    publisher_for = ->(manager : MqttBrokerManager, broker : Model::Broker) {
+      manager.@publishers[broker.id.as(String)]?
+    }
+
+    # A port nothing is listening on, yet
+    free_port = -> {
+      server = TCPServer.new("127.0.0.1", 0)
+      port = server.local_address.port
+      server.close
+      port
+    }
+
+    after_each do
+      managers.each do |manager|
+        manager.stop
+        manager.publishers.each(&.stop)
+      end
+      managers.clear
+      brokers.each { |broker| broker.destroy rescue nil }
+      brokers.clear
+      relays.each(&.cut)
+      relays.clear
+    end
+
+    it "connects to a broker added to the database" do
+      manager = start_manager.call
+      broker = new_broker.call(test_broker.port)
+
+      eventually { publisher_for.call(manager, broker).try { |pub| !pub.client.closed? } }
+    end
+
+    it "reconnects with the new password when it changes" do
+      manager = start_manager.call
+      broker = new_broker.call(test_broker.port)
+      eventually { publisher_for.call(manager, broker) }
+      original = publisher_for.call(manager, broker).not_nil!
+
+      broker.password = "second"
+      broker.save!
+
+      eventually { publisher_for.call(manager, broker).try &.broker.password.== "second" }
+      replacement = publisher_for.call(manager, broker).not_nil!
+      replacement.should_not be original
+      eventually { !replacement.client.closed? }
+      original.client.closed?.should be_true
+    end
+
+    it "keeps the connection when only the name changes" do
+      manager = start_manager.call
+      broker = new_broker.call(test_broker.port)
+      eventually { publisher_for.call(manager, broker) }
+      original = publisher_for.call(manager, broker).not_nil!
+
+      broker.name = "renamed-#{UUID.random}"
+      broker.save!
+
+      eventually { publisher_for.call(manager, broker).try &.broker.name.== broker.name }
+      publisher_for.call(manager, broker).should be original
+      original.client.closed?.should be_false
+    end
+
+    it "moves to a new address once it is reachable" do
+      manager = start_manager.call
+      broker = new_broker.call(test_broker.port)
+      eventually { publisher_for.call(manager, broker) }
+
+      port = free_port.call
+      broker.host = "127.0.0.1"
+      broker.port = port
+      broker.save!
+
+      # the new address is not up yet
+      sleep 2.seconds
+      relays << BrokerRelay.new(test_broker.host, test_broker.port, port)
+
+      eventually(40.seconds) { publisher_for.call(manager, broker).try { |pub| pub.broker.port == port && !pub.client.closed? } }
+    end
+
+    it "republishes state to the broker at a new address" do
+      manager = start_manager.call
+      ready = [] of String
+      manager.on_broker_ready = ->(id : String) { ready << id; nil }
+      broker = new_broker.call(test_broker.port)
+      eventually { publisher_for.call(manager, broker) }
+      ready.clear
+
+      relay = BrokerRelay.new(test_broker.host, test_broker.port)
+      relays << relay
+      broker.host = "127.0.0.1"
+      broker.port = relay.port
+      broker.save!
+
+      eventually { publisher_for.call(manager, broker).try &.broker.port.== relay.port }
+      eventually { ready.includes?(broker.id) }
+    end
+
+    it "ends up with the last of several quick changes" do
+      manager = start_manager.call
+      broker = new_broker.call(test_broker.port)
+      eventually { publisher_for.call(manager, broker) }
+
+      %w[a b c d e].each do |password|
+        broker.password = password
+        broker.save!
+      end
+
+      sleep 3.seconds
+      publisher = publisher_for.call(manager, broker).not_nil!
+      publisher.broker.password.should eq "e"
+      publisher.client.closed?.should be_false
+      manager.publishers.count(&.broker.id.== broker.id).should eq 1
+    end
+
+    it "disconnects from a broker removed from the database" do
+      manager = start_manager.call
+      broker = new_broker.call(test_broker.port)
+      eventually { publisher_for.call(manager, broker) }
+      publisher = publisher_for.call(manager, broker).not_nil!
+
+      broker.destroy
+
+      eventually { publisher_for.call(manager, broker).nil? }
+      eventually { publisher.client.closed? }
+    end
+  end
+end
+
+def eventually(timeout : Time::Span = 15.seconds, &)
+  deadline = Time.instant + timeout
+  until yield
+    raise "timed out after #{timeout}" if Time.instant > deadline
+    sleep 100.milliseconds
+  end
+end

spec/publishing/mqtt_broker_manager_spec.cr

@@ -17,6 +17,11 @@ module PlaceOS::Source
 
       publisher = publisher_manager.@publishers[model.id.as(String)]?
       publisher.should_not be_nil
+    ensure
+      # NOTE:: a manager left running acts on every Broker later specs save,
+      # and its publishers take their sessions over
+      publisher_manager.try &.stop
+      publisher_manager.try &.publishers.each(&.stop)
     end
   end
 end

spec/spec_helper.cr

@@ -222,8 +222,9 @@ module PlaceOS::Source
     @sockets = [] of TCPSocket
     @lock = Mutex.new
 
-    def initialize(@upstream_host : String, @upstream_port : Int32)
-      @server = TCPServer.new("127.0.0.1", 0)
+    # `port` 0 picks a free one
+    def initialize(@upstream_host : String, @upstream_port : Int32, port : Int32 = 0)
+      @server = TCPServer.new("127.0.0.1", port)
       @port = @server.local_address.port
       spawn { accept_loop }
     end

src/source/publishing/mqtt_broker_manager.cr

@@ -57,27 +57,74 @@ module PlaceOS::Source
       hash
     end
 
+    # NOTE:: the event only says which Broker changed. What is applied is the
+    # Broker as it is in the database now, one change per Broker at a time.
+    # Events are processed concurrently, so two edits made close together
+    # could otherwise finish out of order, and the older one win
     def process_resource(action : Resource::Action, resource : Model::Broker) : Resource::Result
-      # Don't recreate the publisher if only "safe" attributes have changed
-      case action
-      in .created? then create_publisher(resource)
-      in .updated?
-        if MqttBrokerManager.safe_update?(resource)
-          update_publisher(resource)
+      broker_id = resource.id.as(String)
+      broker = action.deleted? ? nil : current_broker(resource)
+      apply(broker_id, broker)
+    end
+
+    # The Broker's row in the database, or `nil` once it is gone
+    #
+    # A Broker that was never saved has no row to read, so it is taken as is
+    private def current_broker(resource : Model::Broker) : Model::Broker?
+      return resource if resource.new_record?
+      Model::Broker.find?(resource.id.as(String))
+    rescue e
+      Log.warn(exception: e) { "could not read Broker<#{resource.id}>, applying the change as received" }
+      resource
+    end
+
+    # Brings the publisher for a Broker in line with it: created, replaced,
+    # updated in place or removed
+    protected def apply(broker_id : String, broker : Model::Broker?, retry : Bool = true) : Resource::Result
+      broker_lock(broker_id).synchronize do
+        return remove_publisher(broker_id) if broker.nil?
+
+        existing = read_publishers(&.[broker_id]?)
+        if existing.nil?
+          create_publisher(broker, retry: retry, resync: startup_finished?)
+        elsif MqttBrokerManager.same_connection?(existing.broker, broker)
+          existing.set_broker(broker)
+          Resource::Result::Success
         else
-          # Recreate the publisher
-          create_publisher(resource)
+          # NOTE:: a new connection may be to a different broker altogether,
+          # one that holds none of our state, so publish all of it again
+          create_publisher(broker, retry: retry, resync: true)
         end
-      in .deleted? then remove_publisher(resource)
       end
     end
 
-    # Attributes that can change without recreating the publisher
-    SAFE_ATTRIBUTES = [:name, :description, :filters]
+    # Whether both connect the same way. A change to anything else is applied
+    # to the running publisher, without reconnecting
+    def self.same_connection?(current : Model::Broker, updated : Model::Broker) : Bool
+      current.host == updated.host &&
+        current.port == updated.port &&
+        current.tls == updated.tls &&
+        current.auth_type == updated.auth_type &&
+        current.username == updated.username &&
+        current.password == updated.password &&
+        current.certificate == updated.certificate
+    end
+
+    private getter broker_locks = Hash(String, Mutex).new
+    private getter broker_locks_lock = Mutex.new
+
+    private def broker_lock(broker_id : String) : Mutex
+      broker_locks_lock.synchronize do
+        broker_locks[broker_id] ||= Mutex.new
+      end
+    end
 
-    # Create a `MqttPublisher` for the `Broker`
+    # Create a `MqttPublisher` for the `Broker`, replacing any there is
     #
-    protected def create_publisher(broker : Model::Broker, retry : Bool = true) : Resource::Result
+    # The existing publisher keeps running until its replacement has connected,
+    # so a Broker updated with settings that do not work yet stays connected
+    # with the ones that did
+    private def create_publisher(broker : Model::Broker, retry : Bool, resync : Bool) : Resource::Result
       broker_id = broker.id.as(String)
       existing = read_publishers(&.[broker_id]?)
       existing.try &.superseded = true
@@ -94,13 +141,12 @@ module PlaceOS::Source
         return Resource::Result::Error
       end
 
-      already_publishing = write_publishers do |publishers|
+      write_publishers do |publishers|
         # Close off exisiting publisher, if present
         existing = publishers[broker_id]?
         existing.try &.superseded = true
         existing.stop unless existing.nil?
         publishers[broker_id] = publisher
-        !existing.nil?
       end
 
       publisher.on_reconnect = -> {
@@ -110,15 +156,13 @@ module PlaceOS::Source
       }
       publisher.start
 
-      # Trigger state sync callback if this is a new broker after startup.
-      #
-      # NOTE:: "new" has to mean we had no publisher for it, not merely that
-      # startup has finished. `Resource#start` opens the changefeed before it
-      # loads existing rows, so a create event for a Broker we already loaded
-      # arrives once startup is marked complete — and re-publishing all state
-      # for a Broker we were already connected to is wasted work
-      if startup_finished? && !already_publishing
-        Log.info { "new broker connected after startup, triggering state sync for Broker<#{broker_id}>" }
+      # NOTE:: not for the Brokers loaded at startup, the initial sync covers
+      # them. `Resource#start` opens the changefeed before it loads existing
+      # rows, so a create event for one of those can arrive once startup is
+      # marked complete, but it finds a publisher with the same connection
+      # and goes no further
+      if resync
+        Log.info { "connected to Broker<#{broker_id}>, triggering state sync" }
         on_broker_ready.try &.call(broker_id)
       end
 
@@ -129,7 +173,11 @@ module PlaceOS::Source
     private getter retrying_lock = Mutex.new
 
     # Tries the Broker again with backoff, reading it from the database each
-    # time, until a publisher for it exists or the Broker is gone
+    # time, until a publisher is connected with its current settings or the
+    # Broker is gone
+    #
+    # NOTE:: a publisher existing is not enough. When an update fails to
+    # connect, the publisher it was replacing carries on with the old settings
     protected def retry_publisher(broker_id : String) : Nil
       return unless retrying_lock.synchronize { retrying.add?(broker_id) }
 
@@ -139,40 +187,22 @@ module PlaceOS::Source
           loop do
             sleep delay
             delay = {delay * 2, 30.seconds}.min
-            break if read_publishers(&.has_key?(broker_id))
             break unless broker = Model::Broker.find?(broker_id)
-            break if create_publisher(broker, retry: false).success?
+            current = read_publishers(&.[broker_id]?)
+            break if current && MqttBrokerManager.same_connection?(current.broker, broker)
+            break unless apply(broker_id, broker, retry: false).error?
           end
+        rescue e
+          Log.error(exception: e) { "stopped retrying Broker<#{broker_id}>" }
         ensure
           retrying_lock.synchronize { retrying.delete(broker_id) }
         end
       end
     end
 
-    # Update safe fields on the `MqttPublisher`'s `Broker`
-    #
-    protected def update_publisher(broker : Model::Broker) : Resource::Result
-      broker_id = broker.id.as(String)
-
-      success = write_publishers do |publishers|
-        publisher = publishers[broker_id]?
-        if publisher
-          publisher.set_broker(broker)
-          true
-        else
-          Log.error { "missing existing publisher for Broker<#{broker_id}>" }
-          false
-        end
-      end
-
-      # Create the publisher if the update failed
-      success ? Resource::Result::Success : create_publisher(broker)
-    end
-
     # Close and remove the `MqttPublisher` for the `Broker`
     #
-    private def remove_publisher(broker : Model::Broker) : Resource::Result
-      broker_id = broker.id.as(String)
+    private def remove_publisher(broker_id : String) : Resource::Result
       existing = write_publishers do |publishers|
         publishers.delete(broker_id)
       end
@@ -206,14 +236,5 @@ module PlaceOS::Source
         yield @publishers
       end
     end
-
-    # Safe to update if fields in SAFE_ATTRIBUTES changed
-    #
-    def self.safe_update?(model : Model::Broker)
-      # Take the union of the changed fields and the safe fields
-      attribute_union = model.changed_attributes.keys | SAFE_ATTRIBUTES
-      # Union should be the same size iff fields are in SAFE_ATTRIBUTES
-      attribute_union.size == SAFE_ATTRIBUTES.size
-    end
   end
 end