fix(mqtt): keep reconnecting after a session takeover

shard.lock

@@ -139,7 +139,7 @@ shards:
 
   mqtt:
     git: https://github.com/spider-gazelle/crystal-mqtt.git
-    version: 2.0.1
+    version: 2.0.2
 
   murmur3:
     git: https://github.com/aca-labs/murmur3.git

spec/publishing/influx_ingest_spec.cr

@@ -95,58 +95,6 @@ module PlaceOS::Source
     end
   end
 
-  # Relays TCP to the MQTT broker until it is cut, standing in for a broker
-  # that goes away under a connected publisher. Once cut nothing listens on the
-  # port, so reconnection attempts are refused
-  class BrokerRelay
-    getter port : Int32
-
-    @sockets = [] of TCPSocket
-    @lock = Mutex.new
-
-    def initialize(@upstream_host : String, @upstream_port : Int32)
-      @server = TCPServer.new("127.0.0.1", 0)
-      @port = @server.local_address.port
-      spawn { accept_loop }
-    end
-
-    private def accept_loop
-      while client = @server.accept?
-        upstream = TCPSocket.new(@upstream_host, @upstream_port)
-        @lock.synchronize { @sockets << client << upstream }
-        spawn { pipe(client, upstream) }
-        spawn { pipe(upstream, client) }
-      end
-    end
-
-    private def pipe(from, to)
-      IO.copy(from, to)
-    rescue
-    ensure
-      from.close rescue nil
-      to.close rescue nil
-    end
-
-    def cut : Nil
-      @server.close rescue nil
-      @lock.synchronize do
-        @sockets.each { |socket| socket.close rescue nil }
-        @sockets.clear
-      end
-    end
-
-    def broker : Model::Broker
-      broker = Model::Broker.new(
-        name: "relay-#{UUID.random}",
-        host: "127.0.0.1",
-        port: port,
-        auth_type: :no_auth,
-      )
-      broker.id = "spec-relay-#{UUID.random}"
-      broker
-    end
-  end
-
   def self.influx_manager : InfluxManager
     manager = InfluxManager.new(INFLUX_HOST.not_nil!, INFLUX_API_KEY.not_nil!)
     # rather than the production interval, so specs are not waiting on it

spec/publishing/mqtt_publisher_spec.cr

@@ -31,6 +31,37 @@ module PlaceOS::Source
       end
     end
 
+    # what the broker retains is out of date once the connection drops, so the
+    # reconnection has to be followed by republishing state. That happens once
+    # the broker has accepted the new connection, never ahead of it
+    describe "reconnection" do
+      it "calls on_reconnect once the broker accepts the new connection" do
+        relay = PlaceOS::Source::BrokerRelay.new(test_broker.host, test_broker.port)
+        publisher = MqttPublisher.new(relay.broker)
+        reconnected = Channel(Bool).new(1)
+        publisher.on_reconnect = -> {
+          # nothing would be published from here if the client was not usable
+          reconnected.send(publisher.client.closed?)
+          nil
+        }
+        publisher.start
+
+        begin
+          relay.drop
+
+          select
+          when closed = reconnected.receive
+            closed.should be_false
+          when timeout(15.seconds)
+            fail "on_reconnect was not called"
+          end
+        ensure
+          publisher.stop rescue nil
+          relay.cut
+        end
+      end
+    end
+
     describe "#publish" do
       it "writes a status event to its MQTT topic" do
         client = PlaceOS::Source.subscriber

spec/spec_helper.cr

@@ -211,4 +211,61 @@ module PlaceOS::Source
 
     state
   end
+
+  # Relays TCP to the MQTT broker, standing in for a broker that goes away
+  # under a connected publisher. `drop` breaks the connections open now and
+  # keeps listening; once `cut` nothing listens on the port, so reconnection
+  # attempts are refused
+  class BrokerRelay
+    getter port : Int32
+
+    @sockets = [] of TCPSocket
+    @lock = Mutex.new
+
+    def initialize(@upstream_host : String, @upstream_port : Int32)
+      @server = TCPServer.new("127.0.0.1", 0)
+      @port = @server.local_address.port
+      spawn { accept_loop }
+    end
+
+    private def accept_loop
+      while client = @server.accept?
+        upstream = TCPSocket.new(@upstream_host, @upstream_port)
+        @lock.synchronize { @sockets << client << upstream }
+        spawn { pipe(client, upstream) }
+        spawn { pipe(upstream, client) }
+      end
+    end
+
+    private def pipe(from, to)
+      IO.copy(from, to)
+    rescue
+    ensure
+      from.close rescue nil
+      to.close rescue nil
+    end
+
+    def cut : Nil
+      @server.close rescue nil
+      drop
+    end
+
+    def drop : Nil
+      @lock.synchronize do
+        @sockets.each { |socket| socket.close rescue nil }
+        @sockets.clear
+      end
+    end
+
+    def broker : Model::Broker
+      broker = Model::Broker.new(
+        name: "relay-#{UUID.random}",
+        host: "127.0.0.1",
+        port: port,
+        auth_type: :no_auth,
+      )
+      broker.id = "spec-relay-#{UUID.random}"
+      broker
+    end
+  end
 end

src/source/publishing/mqtt_broker_manager.cr

@@ -79,9 +79,12 @@ module PlaceOS::Source
     #
     protected def create_publisher(broker : Model::Broker, retry : Bool = true) : Resource::Result
       broker_id = broker.id.as(String)
+      existing = read_publishers(&.[broker_id]?)
+      existing.try &.superseded = true
       begin
         publisher = MqttPublisher.new(broker)
       rescue e
+        existing.try &.superseded = false
         # NOTE:: any error, not only `IO::Error`: a refused connection surfaces
         # as `MQTT::NotConnectedError`, and an unhandled one drops the Broker
         # for good — nothing is published to it until source restarts. A broker
@@ -94,6 +97,7 @@ module PlaceOS::Source
       already_publishing = 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?

src/source/publishing/mqtt_publisher.cr

@@ -39,28 +39,32 @@ module PlaceOS::Source
     # Called once the client has re-established a dropped connection
     property on_reconnect : Proc(Nil)? = nil
 
-    @connections = Atomic(Int32).new(0)
+    @stopping = false
+
+    # Set while a publisher replacing this one connects. Both use the broker id
+    # as the client id, so the broker hands the session over and ends ours
+    property? superseded : Bool = false
 
     def initialize(@broker : PlaceOS::Model::Broker)
-      @client = MqttPublisher.client(@broker) { connection_opened }
+      @client = MqttPublisher.client(@broker)
+
+      # NOTE:: a broker that restarts without persistence comes back empty, and
+      # one that was down missed whatever changed meanwhile. Either way what it
+      # retains is no longer the state of the system, so a reconnection has to
+      # be followed by publishing all of it again. The client calls this once
+      # the broker has accepted the new connection, and from its own fiber
+      @client.on_reconnect { |_session_present| on_reconnect.try &.call }
+
+      spawn { watch_connection }
     end
 
-    # NOTE:: a broker that restarts without persistence comes back empty, and
-    # one that was down missed whatever changed meanwhile. Either way what it
-    # retains is no longer the state of the system, so a reconnection has to be
-    # followed by publishing all of it again. The client has no hook for that,
-    # but it builds a new transport for every connection, so the second and
-    # later transports mark reconnections
-    private def connection_opened : Nil
-      return if @connections.add(1) == 0
-      callback = on_reconnect
-      return unless callback
-
-      # after the client has resumed its session on the new transport
-      spawn do
-        sleep 1.second
-        callback.call
-      end
+    # The client reconnects by itself, and only stops for good when the broker
+    # says not to come back: credentials it refuses, or another client taking
+    # over our client id. Nothing is published to it after that, so say so
+    private def watch_connection : Nil
+      client.wait_close
+      return if @stopping || superseded?
+      Log.error { "MQTT client for Broker<#{broker.id}> on #{broker.host}:#{broker.port} has stopped and will not reconnect" }
     end
 
     # The protocol version that was actually negotiated with this broker
@@ -69,6 +73,7 @@ module PlaceOS::Source
     end
 
     def stop
+      @stopping = true
       super
       # NOTE:: disconnect first. `wait_close` blocks until the connection has
       # terminated, so waiting before asking it to stop hangs until the broker
@@ -90,7 +95,7 @@ module PlaceOS::Source
     # The connection settings are captured here rather than read from the model
     # on each attempt: a `Broker` update that changes them is not a "safe"
     # update, so `MqttBrokerManager` replaces the publisher outright
-    def self.client(broker : PlaceOS::Model::Broker, &on_transport : -> Nil) : ::MQTT::Client
+    def self.client(broker : PlaceOS::Model::Broker) : ::MQTT::Client
       Log.debug { {message: "creating MQTT client", host: broker.host, port: broker.port.to_s, tls: broker.tls} }
 
       host = broker.host
@@ -101,7 +106,6 @@ module PlaceOS::Source
       # be reopened. This is what lets the client re-establish the connection
       # itself, and what version negotiation needs for its fallback attempt
       client = ::MQTT::Client.new(reconnect: ::MQTT::Reconnect.new) do
-        on_transport.call
         tls = if secure
                 tls_client = OpenSSL::SSL::Context::Client.new
                 tls_client.verify_mode = OpenSSL::SSL::VerifyMode::NONE
@@ -156,7 +160,11 @@ module PlaceOS::Source
             Log.warn(exception: e) { "MQTT publish failed, waiting for the client to reconnect" }
           }
         ) do
-          # NOTE:: nothing waits on a broker that is down. Retrying every
+          # NOTE:: `closed?` holds until the broker has accepted a connection,
+          # not only while the socket is down: publishing ahead of that is a
+          # protocol error to a broker still taking an old session over.
+          #
+          # Nothing waits on a broker that is down. Retrying every
           # message until it returned held this queue for half a minute each,
           # which filled it and blocked `broadcast` — and with it every other
           # sink, InfluxDB included. Reconnecting republishes all state