From f0b0917cb4609a003d786b8f8b8ba0c150b4eaf4 Mon Sep 17 00:00:00 2001 From: Stephen von Takach Date: Tue, 27 Oct 2026 23:13:51 +1100 Subject: [PATCH] fix(mqtt): keep reconnecting after a session takeover Bump mqtt to 2.0.2, which keeps retrying when a reconnection drops straight away or the broker reports a protocol error, and holds application packets until the broker accepts the connection. Resync from the client's on_reconnect callback, once the broker has accepted the new connection, instead of counting transports. Log an error when the client stops for good, except when the manager replaces a publisher. Co-Authored-By: Claude Opus 5.5 --- shard.lock | 2 +- spec/publishing/influx_ingest_spec.cr | 52 ------------------ spec/publishing/mqtt_publisher_spec.cr | 31 +++++++++++ spec/spec_helper.cr | 57 ++++++++++++++++++++ src/source/publishing/mqtt_broker_manager.cr | 4 ++ src/source/publishing/mqtt_publisher.cr | 50 +++++++++-------- 6 files changed, 122 insertions(+), 74 deletions(-) diff --git a/shard.lock b/shard.lock index 19e7ca4..08f3ca6 100644 --- a/shard.lock +++ b/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 diff --git a/spec/publishing/influx_ingest_spec.cr b/spec/publishing/influx_ingest_spec.cr index 5d55217..fc30b85 100644 --- a/spec/publishing/influx_ingest_spec.cr +++ b/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 diff --git a/spec/publishing/mqtt_publisher_spec.cr b/spec/publishing/mqtt_publisher_spec.cr index 409b8c7..074f398 100644 --- a/spec/publishing/mqtt_publisher_spec.cr +++ b/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 diff --git a/spec/spec_helper.cr b/spec/spec_helper.cr index ec660ba..6fa835c 100644 --- a/spec/spec_helper.cr +++ b/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 diff --git a/src/source/publishing/mqtt_broker_manager.cr b/src/source/publishing/mqtt_broker_manager.cr index f672cb9..e94ca80 100644 --- a/src/source/publishing/mqtt_broker_manager.cr +++ b/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? diff --git a/src/source/publishing/mqtt_publisher.cr b/src/source/publishing/mqtt_publisher.cr index d6b79bc..165aeda 100644 --- a/src/source/publishing/mqtt_publisher.cr +++ b/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