feat(api): clear and resync the MQTT brokers on request

spec/clear_spec.cr

@@ -0,0 +1,37 @@
+require "./spec_helper"
+
+module PlaceOS::Source
+  describe "clearing retained state" do
+    # several rounds, as a broker that drops part of a large subscription needs
+    it "removes every state and metadata topic, in rounds, and nothing else" do
+      with_publisher do |publisher|
+        publisher.start
+        manager = MqttBrokerManager.new
+        manager.@publishers[publisher.broker.id.as(String)] = publisher
+        mappings = Mappings.new
+        event, _ = unique_status_event
+        mappings.published.add(event)
+        events = StatusEvents.new(mappings, [manager] of PublisherManager)
+
+        org = "org-#{UUID.random}"
+        state = Array.new(45) { |i| "placeos/#{org}/state/b/l/a/cs-1/driver-1/Display/#{i}/power" }
+        metadata = Array.new(5) { |i| "placeos/#{org}/metadata/sys-#{i}" }
+        other = "elsewhere/#{org}/keep"
+        client = subscriber
+        (state + metadata + [other]).each { |topic| client.publish(topic, %({"value":true}), retain: true) }
+
+        mine = ->(topic : String) { topic.includes?(org) }
+        publisher.retained_topics("placeos/#{org}/#", &mine).size.should eq 50
+
+        events.clear_retained(batch: 20).should be >= 50
+
+        publisher.retained_topics("placeos/#{org}/#", &mine).should be_empty
+        publisher.retained_topics("elsewhere/#{org}/#", &mine).should eq Set{other}
+        mappings.published.for_module(event.module_id).should be_empty
+
+        client.publish(other, "", retain: true)
+        client.disconnect
+      end
+    end
+  end
+end

spec/manager_spec.cr

@@ -49,5 +49,20 @@ module PlaceOS::Source
 
       manager.stop
     end
+
+    describe "resync" do
+      it "runs one at a time" do
+        manager = Manager.new([] of PublisherManager)
+        manager.resync_status.state.idle?.should be_true
+
+        # never started, so it waits on the initial sync and stays running
+        manager.resync(clear: true).should be_true
+        manager.resync.should be_false
+        status = manager.resync_status
+        status.state.clearing?.should be_true
+        status.clear?.should be_true
+        status.started_at.should_not be_nil
+      end
+    end
   end
 end

src/controllers/root.cr

@@ -31,6 +31,30 @@ module PlaceOS::Source::Api
       )
     end
 
+    # Publishes everything to the MQTT brokers again: every module's state and
+    # every model's metadata. With `clear`, everything source publishes is
+    # removed from the brokers first, for when what they hold has drifted.
+    #
+    # Runs in the background: 202 once started, 409 while one is running.
+    # Progress is at `GET /resync`
+    @[AC::Route::POST("/resync", status_code: HTTP::Status::ACCEPTED)]
+    def resync(
+      @[AC::Param::Info(description: "remove everything source publishes from the brokers first", example: "true")]
+      clear : Bool = false,
+    ) : Manager::ResyncStatus
+      manager = Manager.instance
+      unless manager.resync(clear)
+        render status: :conflict, json: manager.resync_status
+      end
+      manager.resync_status
+    end
+
+    # Where the last resync got to
+    @[AC::Route::GET("/resync")]
+    def resync_status : Manager::ResyncStatus
+      Manager.instance.resync_status
+    end
+
     def self.healthcheck : NamedTuple(healthy: Bool, services: Hash(String, NamedTuple(status: String, error: String?)))
       results = Promise.all(
         Promise.defer {

src/source/manager.cr

@@ -70,6 +70,113 @@ module PlaceOS::Source
       Fiber.yield
     end
 
+    # Resync
+    ###########################################################################
+
+    # Where the last resync got to
+    struct ResyncStatus
+      include JSON::Serializable
+
+      enum State
+        Idle
+        Clearing
+        Resyncing
+        Complete
+        Failed
+      end
+
+      getter state : State
+      # whether everything retained was removed first
+      getter? clear : Bool
+      getter started_at : Time?
+      getter finished_at : Time?
+      # retained topics removed, when clearing
+      getter cleared : Int64
+      # models whose metadata was published again
+      getter metadata : Int32
+      getter error : String?
+
+      def initialize(@state = State::Idle, @clear = false, @started_at = nil, @finished_at = nil, @cleared = 0_i64, @metadata = 0, @error = nil)
+      end
+
+      def running? : Bool
+        state.clearing? || state.resyncing?
+      end
+
+      def copy_with(**changes) : self
+        ResyncStatus.new(**{
+          state:       @state,
+          clear:       @clear,
+          started_at:  @started_at,
+          finished_at: @finished_at,
+          cleared:     @cleared,
+          metadata:    @metadata,
+          error:       @error,
+        }.merge(changes))
+      end
+    end
+
+    private getter resync_lock = Mutex.new
+    @resync_status = ResyncStatus.new
+
+    def resync_status : ResyncStatus
+      resync_lock.synchronize { @resync_status }
+    end
+
+    private def resync_status=(status : ResyncStatus)
+      resync_lock.synchronize { @resync_status = status }
+    end
+
+    # Publishes everything to the MQTT brokers again: every Module's state and
+    # every model's metadata. With `clear`, everything source publishes is
+    # removed from the brokers first, for when what they hold has drifted.
+    #
+    # Runs in the background, one at a time. Returns false if one is running.
+    def resync(clear : Bool = false) : Bool
+      resync_lock.synchronize do
+        return false if @resync_status.running?
+        @resync_status = ResyncStatus.new(
+          state: clear ? ResyncStatus::State::Clearing : ResyncStatus::State::Resyncing,
+          clear: clear,
+          started_at: Time.utc,
+        )
+      end
+      spawn { run_resync(clear) }
+      true
+    end
+
+    private def run_resync(clear : Bool) : Nil
+      # what is resynced is the state the initial sync has read
+      until status_events.initial_sync_complete?
+        sleep 1.second
+      end
+
+      Log.info { {message: "resyncing MQTT brokers", clear: clear} }
+      if clear
+        cleared = status_events.clear_retained
+        self.resync_status = resync_status.copy_with(state: ResyncStatus::State::Resyncing, cleared: cleared)
+      end
+
+      metadata = zone_router.republish_metadata +
+                 control_system_router.republish_metadata +
+                 driver_router.republish_metadata
+      status_events.resync_state
+
+      self.resync_status = resync_status.copy_with(
+        state: ResyncStatus::State::Complete,
+        metadata: metadata,
+        finished_at: Time.utc,
+      )
+      Log.info { {message: "resynced MQTT brokers", clear: clear, metadata: metadata.to_s} }
+    rescue error
+      Log.error(exception: error) { "resyncing MQTT brokers" }
+      self.resync_status = resync_status.copy_with(
+        state: ResyncStatus::State::Failed,
+        error: error.message || error.class.name,
+        finished_at: Time.utc,
+      )
+    end
+
     def stop
       return unless started?
 

src/source/mappings.cr

@@ -141,6 +141,25 @@ module PlaceOS::Source
 
     alias Data = Status | Metadata
 
+    # Every row of a model, a page at a time, in primary key order
+    def self.each_model(model : T.class, page_size : Int32 = 64, & : T ->) forall T
+      last_id = nil
+      loop do
+        query = model.order(id: :asc)
+        query = query.where("id > ?", last_id) unless last_id.nil?
+        page = query.limit(page_size).to_a
+        page.each { |row| yield row }
+        break if page.size < page_size
+        last_id = page.last.id
+        break if last_id.nil?
+      end
+    end
+
+    # Systems with a top-level zone, whose models are published
+    def scoped_system?(control_system_id : String) : Bool
+      read { |state| !!state.system_zones[control_system_id]?.try(&.has_key?(Mappings.scope)) }
+    end
+
     # Hierarchy and Scoping
     ###########################################################################
 

src/source/published.cr

@@ -44,6 +44,14 @@ module PlaceOS::Source
       end
     end
 
+    # Forget everything, once nothing is retained any more
+    def clear : Nil
+      lock.synchronize do
+        events.clear
+        by_system.clear
+      end
+    end
+
     def for_module(module_id : String) : Set(Mappings::Status)
       lock.synchronize { events[module_id]?.try(&.dup) || Set(Mappings::Status).new }
     end

src/source/publishing/mqtt_publisher.cr

@@ -199,22 +199,37 @@
     # The status topics the broker currently retains, as a client connecting
     # now would be handed them, that `keep` returns true for. Each topic is
     # passed to `keep` as it arrives, so only those wanted are held.
+    def retained_state_topics(quiet : Time::Span = 2.seconds, limit : Time::Span = 2.minutes, &keep : String -> Bool) : Set(String)
+      retained_topics(File.join(MQTT_NAMESPACE, "+", "state", "#"), quiet, limit, &keep)
+    end
+
+    # The topics under `filter` the broker currently retains, that `keep`
+    # returns true for, up to `max` of them.
     #
     # The broker sends retained messages straight after it acknowledges the
-    # subscription, so collection ends once `quiet` passes without one, or at
-    # `limit` regardless. Only retained messages count: a live publication
-    # arriving meanwhile is not part of what was there.
-    def retained_state_topics(quiet : Time::Span = 2.seconds, limit : Time::Span = 2.minutes, &keep : String -> Bool) : Set(String)
-      filter = File.join(MQTT_NAMESPACE, "+", "state", "#")
+    # subscription, so collection ends once `quiet` passes without one, at
+    # `limit`, or once `max` are held. Only retained messages count: a live
+    # publication arriving meanwhile is not part of what was there.
+    #
+    # NOTE:: the broker drops what a client's outgoing queue cannot hold
+    # (`max_queued_messages`), so on a large broker this can be part of what is
+    # retained. Repeat it to see the rest once those found are dealt with
+    def retained_topics(
+      filter : String,
+      quiet : Time::Span = 2.seconds,
+      limit : Time::Span = 2.minutes,
+      max : Int32 = Int32::MAX,
+      &keep : String -> Bool
+    ) : Set(String)
       topics = Set(String).new
       lock = Mutex.new
       last = Time.instant
 
       client.subscribe(filter) do |topic, payload, retained|
         next unless retained && !payload.empty?
         wanted = keep.call(topic)
         lock.synchronize do
-          topics << topic if wanted
+          topics << topic if wanted && topics.size < max
           last = Time.instant
         end
       end
@@ -224,7 +239,7 @@
         loop do
           sleep 100.milliseconds
           now = Time.instant
-          break if lock.synchronize { now - last >= quiet } || now - started >= limit
+          break if lock.synchronize { now - last >= quiet || topics.size >= max } || now - started >= limit
         end
       ensure
         client.unsubscribe(filter) rescue nil

src/source/router/control_system_router.cr

@@ -86,7 +86,7 @@
           mappings.write { |state| state.system_zones[control_system_id] = zones }
           mappings.set_system_modules(control_system_id, modules)
         else
           mappings.write { |state| state.system_zones.delete(control_system_id) }
           mappings.remove_system_modules(control_system_id)
         end
 
@@ -96,6 +96,18 @@
       Resource::Result::Success
     end
 
+    # Publishes the metadata of every system in scope again. Scope is read
+    # from the mappings, which hold every system's hierarchy
+    def republish_metadata : Int32
+      count = 0
+      Mappings.each_model(PlaceOS::Model::ControlSystem) do |control_system|
+        next unless mappings.scoped_system?(control_system.id.as(String))
+        publish_metadata(control_system)
+        count += 1
+      end
+      count
+    end
+
     def process_resource(action : Resource::Action, resource : PlaceOS::Model::ControlSystem) : Resource::Result
       control_system = resource
 

src/source/router/driver_router.cr

@@ -62,21 +62,38 @@ module PlaceOS::Source::Router
     @loaded_in_scope : Set(String)? = nil
     private getter loaded_in_scope_lock = Mutex.new
 
+    # Publishes the metadata of every Driver in scope again
+    def republish_metadata : Int32
+      scoped = drivers_in_scope
+      count = 0
+      Mappings.each_model(PlaceOS::Model::Driver) do |driver|
+        next unless scoped.includes?(driver.id.as(String))
+        publish_metadata(driver)
+        count += 1
+      end
+      count
+    end
+
     # The Drivers with a Module in a system under a top-level zone, from the
     # mappings, which hold every system and Module by the time Drivers load.
     #
     # Worked out once, for loading. Asking the database meant reading every
     # Module of each Driver, then each Module's systems and their zones
     private def loaded_in_scope : Set(String)
       loaded_in_scope_lock.synchronize do
-        @loaded_in_scope ||= mappings.read do |state|
-          state.drivers.each_with_object(Set(String).new) do |(module_id, driver_id), found|
-            next if found.includes?(driver_id)
-            scoped = state.system_modules[module_id]?.try &.any? do |mapping|
-              state.system_zones[mapping[:control_system_id]]?.try(&.has_key?(Mappings.scope))
-            end
-            found << driver_id if scoped
+        @loaded_in_scope ||= drivers_in_scope
+      end
+    end
+
+    # Drivers with a Module in a system under a top-level zone
+    private def drivers_in_scope : Set(String)
+      mappings.read do |state|
+        state.drivers.each_with_object(Set(String).new) do |(module_id, driver_id), found|
+          next if found.includes?(driver_id)
+          scoped = state.system_modules[module_id]?.try &.any? do |mapping|
+            state.system_zones[mapping[:control_system_id]]?.try(&.has_key?(Mappings.scope))
           end
+          found << driver_id if scoped
         end
       end
     end

src/source/router/zone_router.cr

@@ -47,6 +47,18 @@ module PlaceOS::Source::Router
       end
     end
 
+    # Publishes the metadata of every Zone in scope again
+    def republish_metadata : Int32
+      count = 0
+      Mappings.each_model(PlaceOS::Model::Zone) do |zone|
+        hierarchy_zones = Mappings.hierarchy_zones(zone)
+        next if hierarchy_zones.empty?
+        publish_metadata(hierarchy_zones, zone)
+        count += 1
+      end
+      count
+    end
+
     # Handle a change to a Zone's tags, or its removal.
     #
     # Only systems that have the zone are touched. Every system's mapping used