diff --git a/spec/clear_spec.cr b/spec/clear_spec.cr
new file mode 100644
index 0000000..4a68d88
--- /dev/null
+++ b/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
diff --git a/spec/manager_spec.cr b/spec/manager_spec.cr
index 2c67842..2384dba 100644
--- a/spec/manager_spec.cr
+++ b/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
diff --git a/src/controllers/root.cr b/src/controllers/root.cr
index 4b37867..b3b42f3 100644
--- a/src/controllers/root.cr
+++ b/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 {
diff --git a/src/source/manager.cr b/src/source/manager.cr
index 62127b2..244b3ad 100644
--- a/src/source/manager.cr
+++ b/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?
 
diff --git a/src/source/mappings.cr b/src/source/mappings.cr
index 7123a64..06a4e80 100644
--- a/src/source/mappings.cr
+++ b/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
     ###########################################################################
 
diff --git a/src/source/published.cr b/src/source/published.cr
index 096581c..8686ed7 100644
--- a/src/source/published.cr
+++ b/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
diff --git a/src/source/publishing/mqtt_publisher.cr b/src/source/publishing/mqtt_publisher.cr
index c47e2ae..cb13f89 100644
--- a/src/source/publishing/mqtt_publisher.cr
+++ b/src/source/publishing/mqtt_publisher.cr
@@ -199,13 +199,28 @@ module PlaceOS::Source
     # 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
@@ -214,7 +229,7 @@ module PlaceOS::Source
         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 @@ module PlaceOS::Source
         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
diff --git a/src/source/router/control_system_router.cr b/src/source/router/control_system_router.cr
index 6df1c37..2c0f408 100644
--- a/src/source/router/control_system_router.cr
+++ b/src/source/router/control_system_router.cr
@@ -96,6 +96,18 @@ module PlaceOS::Source::Router
       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
 
diff --git a/src/source/router/driver_router.cr b/src/source/router/driver_router.cr
index cf89ccf..b47c341 100644
--- a/src/source/router/driver_router.cr
+++ b/src/source/router/driver_router.cr
@@ -62,6 +62,18 @@ 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.
     #
@@ -69,14 +81,19 @@ module PlaceOS::Source::Router
     # 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
diff --git a/src/source/router/zone_router.cr b/src/source/router/zone_router.cr
index 21ff71a..1489790 100644
--- a/src/source/router/zone_router.cr
+++ b/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
diff --git a/src/source/status_events.cr b/src/source/status_events.cr
index 6f8d49d..8fdfdf0 100644
--- a/src/source/status_events.cr
+++ b/src/source/status_events.cr
@@ -23,7 +23,10 @@ module PlaceOS::Source
     private getter publisher_managers : Array(PublisherManager)
 
     private property? stopped : Bool = true
-    private property? initial_sync_complete : Bool = false
+    getter? initial_sync_complete : Bool = false
+
+    private def initial_sync_complete=(@initial_sync_complete : Bool)
+    end
 
     private getter sync_lock = Mutex.new(:reentrant)
 
@@ -145,7 +148,7 @@ module PlaceOS::Source
       # delete would then remove a value that is still live
       await_deletions
 
-      Log.info { "resyncing state for new broker connection" }
+      Log.info { "resyncing state" }
 
       mods_mapped = 0_u64
       status_updated = 0_u64
@@ -201,21 +204,27 @@ module PlaceOS::Source
       end
     end
 
+    # Held for each step of processing, so `clear_retained` can hold
+    # processing back: what it clears must not be published again meanwhile
+    private getter process_lock = Mutex.new
+
     protected def process_events
       until stopped?
         begin
-          # Mapping changes first: a batch built after them must be published
-          # against what is retained once they are reconciled
-          reconcile_now(mappings.take_reconcile) if mappings.reconcile_pending?
-          if sweep = take_sweep
-            sweep_now(sweep[0], sweep[1], sweep[0].broker.id.to_s)
+          batch = process_lock.synchronize do
+            # Mapping changes first: a batch built after them must be published
+            # against what is retained once they are reconciled
+            reconcile_now(mappings.take_reconcile) if mappings.reconcile_pending?
+            if sweep = take_sweep
+              sweep_now(sweep[0], sweep[1], sweep[0].broker.id.to_s)
+            end
+
+            build_batch.tap { |events| process_batch(events) unless events.empty? }
           end
 
-          batch = build_batch
           if batch.empty?
             sleep PROCESSING_INTERVAL
           else
-            process_batch(batch)
             Fiber.yield
 
             # This outputs how many writes have occured for each publisher
@@ -459,6 +468,49 @@ module PlaceOS::Source
       keys.includes?(topic)
     end
 
+    # Clearing
+    ###########################################################################
+
+    # Topics a round of `clear_retained` takes from a broker. Below the
+    # broker's queue (`max_queued_messages`) so a round is not cut short by it
+    CLEAR_BATCH = 5_000
+
+    # Removes everything source publishes from every MQTT broker: all `state`
+    # and `metadata` topics, whether or not anything maps to them. Returns how
+    # many topics were removed.
+    #
+    # Event processing is held back throughout, so nothing is published again
+    # while it runs; events that arrive wait, and are processed afterwards.
+    # Nothing is retained once it returns, so a resync must follow it.
+    #
+    # NOTE:: in rounds. A round takes up to `CLEAR_BATCH` topics, removes them
+    # and waits for the removals to reach the broker, so the next subscription
+    # is handed what is left. It ends when a round finds nothing, so it is
+    # complete even when the broker drops part of what it sends a client
+    def clear_retained(batch : Int32 = CLEAR_BATCH) : Int64
+      cleared = 0_i64
+      process_lock.synchronize do
+        publisher_managers.each do |manager|
+          next unless manager.is_a?(MqttBrokerManager)
+          manager.publishers.each do |publisher|
+            {"state", "metadata"}.each do |kind|
+              filter = File.join(MQTT_NAMESPACE, "+", kind, "#")
+              loop do
+                topics = publisher.retained_topics(filter, max: batch) { true }
+                break if topics.empty?
+                topics.each { |topic| publisher.queue_delete_topic(topic) }
+                await_deletions
+                cleared += topics.size
+              end
+            end
+          end
+        end
+        mappings.published.clear
+      end
+      Log.info { {message: "cleared retained state and metadata", removed: cleared.to_s} }
+      cleared
+    end
+
     def self.parse_channel(channel : String) : {String, String}
       _, module_id, status = channel.split('/')
       {module_id, status}