diff --git a/spec/publishing/publisher_spec.cr b/spec/publishing/publisher_spec.cr
index 579a3e4..84b3fcf 100644
--- a/spec/publishing/publisher_spec.cr
+++ b/spec/publishing/publisher_spec.cr
@@ -13,28 +13,35 @@ module PlaceOS::Source
     def delete(message : Publisher::Message)
       handled << "delete:#{message.payload}"
     end
+
+    def delete_topic(topic : String)
+      handled << "delete-topic:#{topic}"
+    end
   end
 
   describe Publisher do
-    # A rename removes the old topic and republishes under the new one. If a
-    # publish were to win that race the delete would take the new value straight
-    # back out, so deletions have to be drained first
-    it "drains queued deletions before queued messages" do
+    # One queue for both, so a deletion can never overtake a stale value
+    # queued before it, nor be overtaken by a value queued after it
+    it "works through publications and deletions in the order they were queued" do
       publisher = RecordingPublisher.new
       event = Mappings::Metadata.new("mod-1234", "hello")
 
-      # queued messages first, deletions second — order of arrival must not
-      # decide order of work
-      3.times { |i| publisher.message_queue.send(Publisher::Message.new(event, "message-#{i}", Time.utc)) }
-      2.times { |i| publisher.queue_delete(Publisher::Message.new(event, "deletion-#{i}", Time.utc)) }
+      publisher.message_queue.send(Publisher::Message.new(event, "message-0", Time.utc))
+      publisher.queue_delete(Publisher::Message.new(event, "deletion-0", Time.utc))
+      publisher.message_queue.send(Publisher::Message.new(event, "message-1", Time.utc))
+      publisher.queue_delete_topic("placeos/org/state/stale")
 
       publisher.start
       sleep 200.milliseconds
 
-      publisher.handled.first(2).should eq ["delete:deletion-0", "delete:deletion-1"]
-      publisher.handled.size.should eq 5
+      publisher.handled.should eq [
+        "publish:message-0",
+        "delete:deletion-0",
+        "publish:message-1",
+        "delete-topic:placeos/org/state/stale",
+      ]
       publisher.deleted.should eq 2
-      publisher.processed.should eq 3
+      publisher.processed.should eq 2
       publisher.stop
     end
 
@@ -53,7 +60,7 @@ module PlaceOS::Source
     end
 
     # a consumer must notice a closed queue rather than spinning on it
-    it "stops cleanly once the queues are closed" do
+    it "stops cleanly once the queue is closed" do
       publisher = RecordingPublisher.new
       publisher.start
       sleep 50.milliseconds
@@ -67,7 +74,6 @@ module PlaceOS::Source
       select
       when done.receive
         publisher.message_queue.closed?.should be_true
-        publisher.delete_queue.closed?.should be_true
       when timeout(5.seconds)
         fail "stop did not return"
       end
diff --git a/spec/publishing/rename_spec.cr b/spec/publishing/rename_spec.cr
index 1288d15..24a37f5 100644
--- a/spec/publishing/rename_spec.cr
+++ b/spec/publishing/rename_spec.cr
@@ -38,26 +38,22 @@ module PlaceOS::Source
     broker = PlaceOS::Source.isolated_broker
     publisher = MqttPublisher.new(broker)
     manager = MockBrokerManager.new(publisher)
-    router = Router::Module.new(mappings, [manager] of PublisherManager)
+    status_events = StatusEvents.new(mappings, [manager] of PublisherManager)
 
     begin
       publisher.start
 
-      # state as it was, under the old path
-      publisher.publish(Publisher::Message.new(
-        mappings.status_events?(module_id, status).not_nil!.first, "true", Time.utc))
+      # state as it was, under the old path: reconciling a Module with nothing
+      # published yet publishes what its storage holds
+      status_events.reconcile_now(Set{module_id})
       sleep 300.milliseconds
 
-      # the rename: clear what is there, then rewrite the mapping
-      router.remove_retained_state(module_id)
+      # the rename rewrites the mapping, then reconciles
       mappings.set_system_modules("cs-1", {
         module_id => {name: new_name, control_system_id: "cs-1", index: new_index},
       })
       new_topic = topic_for(mappings, module_id, status)
-
-      # republish at the new path, which is what a resync does
-      publisher.message_queue.send(Publisher::Message.new(
-        mappings.status_events?(module_id, status).not_nil!.first, "true", Time.utc))
+      status_events.reconcile_now(Set{module_id})
       sleep 800.milliseconds
 
       {old_topic, new_topic}
diff --git a/spec/reconcile_spec.cr b/spec/reconcile_spec.cr
new file mode 100644
index 0000000..2075777
--- /dev/null
+++ b/spec/reconcile_spec.cr
@@ -0,0 +1,162 @@
+require "./spec_helper"
+require "placeos-driver/storage"
+
+module PlaceOS::Source
+  # A system `cs-1` in a resolvable hierarchy, with the given Modules in order.
+  # Modules of the same name are numbered by position, as source does it
+  def self.reconcile_mappings(org : String, modules : Array({String, String})) : Mappings
+    state = Mappings::State.new
+    state.system_zones["cs-1"] = {"org" => org, "building" => "b", "level" => "l", "area" => "a"}
+    mappings = Mappings.new(state)
+    place(mappings, modules)
+    modules.each { |(module_id, _)| state.drivers[module_id] = "driver-1" }
+    mappings
+  end
+
+  def self.place(mappings : Mappings, modules : Array({String, String})) : Nil
+    counts = Hash(String, Int32).new(0)
+    mapping = modules.each_with_object({} of String => Mappings::SystemModule) do |(module_id, name), hash|
+      hash[module_id] = {name: name, control_system_id: "cs-1", index: counts[name] += 1}
+    end
+    mappings.set_system_modules("cs-1", mapping)
+  end
+
+  # what each queued message would publish to, and its payload
+  def self.keyed(messages : Array(Publisher::Message)) : Hash(String, String?)
+    messages.to_h { |m| {MqttPublisher.generate_key(m.data).not_nil!.split('/')[-3..].join('/'), m.payload} }
+  end
+
+  def self.stored(module_id : String, values : Hash(String, String)) : PlaceOS::Driver::RedisStorage
+    store = PlaceOS::Driver::RedisStorage.new(module_id)
+    store.clear
+    values.each { |key, value| store[key] = value }
+    store
+  end
+
+  describe "reconciling retained state" do
+    it "publishes the current state of a Module it has not published" do
+      a = "mod-#{UUID.random}"
+      store = stored(a, {"connected" => "true", "power" => "false"})
+      mappings = reconcile_mappings("org-#{UUID.random}", [{a, "Display"}])
+      manager = MockManager.new
+      events = StatusEvents.new(mappings, [manager] of PublisherManager)
+
+      events.reconcile_now(Set{a})
+
+      keyed(manager.messages).should eq({"Display/1/connected" => "true", "Display/1/power" => "false"})
+      manager.deletions.should be_empty
+      mappings.published.for_module(a).size.should eq 2
+    ensure
+      store.try &.clear
+    end
+
+    it "moves each Module's state to its new index when a system is reordered" do
+      a, b = "mod-#{UUID.random}", "mod-#{UUID.random}"
+      store_a = stored(a, {"which" => %("a")})
+      store_b = stored(b, {"which" => %("b")})
+      mappings = reconcile_mappings("org-#{UUID.random}", [{a, "Display"}, {b, "Display"}])
+      manager = MockManager.new
+      events = StatusEvents.new(mappings, [manager] of PublisherManager)
+      events.reconcile_now(Set{a, b})
+      manager.messages.clear
+
+      place(mappings, [{b, "Display"}, {a, "Display"}])
+      events.reconcile_now(mappings.take_reconcile | Set{a, b})
+
+      keyed(manager.messages).should eq({"Display/1/which" => %("b"), "Display/2/which" => %("a")})
+      # both old topics are taken over by the other Module, so neither is removed
+      manager.deletions.should be_empty
+    ensure
+      store_a.try &.clear
+      store_b.try &.clear
+    end
+
+    it "removes the topics of a Module taken out of a system" do
+      a, b = "mod-#{UUID.random}", "mod-#{UUID.random}"
+      store_a = stored(a, {"connected" => "false"})
+      store_b = stored(b, {"connected" => "true"})
+      mappings = reconcile_mappings("org-#{UUID.random}", [{a, "Display"}, {b, "Display"}])
+      manager = MockManager.new
+      events = StatusEvents.new(mappings, [manager] of PublisherManager)
+      events.reconcile_now(Set{a, b})
+      manager.messages.clear
+
+      # b moves up to Display_1, which a held; Display_2 is left with nothing
+      place(mappings, [{b, "Display"}])
+      events.reconcile_now(Set{a, b})
+
+      keyed(manager.messages).should eq({"Display/1/connected" => "true"})
+      keyed(manager.deletions).keys.should eq ["Display/2/connected"]
+      mappings.published.for_module(a).should be_empty
+    ensure
+      store_a.try &.clear
+      store_b.try &.clear
+    end
+
+    it "removes a deleted Module's topics even once its storage is gone" do
+      a = "mod-#{UUID.random}"
+      store = stored(a, {"connected" => "false", "power" => "true"})
+      mappings = reconcile_mappings("org-#{UUID.random}", [{a, "Display"}])
+      manager = MockManager.new
+      events = StatusEvents.new(mappings, [manager] of PublisherManager)
+      events.reconcile_now(Set{a})
+
+      # core clears the storage, and the system drops the Module, before the
+      # Module's own delete event is seen
+      store.clear
+      place(mappings, [] of {String, String})
+      events.reconcile_now(Set{a})
+
+      keyed(manager.deletions).keys.sort.should eq ["Display/1/connected", "Display/1/power"]
+      mappings.published.for_module(a).should be_empty
+    ensure
+      store.try &.clear
+    end
+
+    it "removes the topics of a system that is gone" do
+      a = "mod-#{UUID.random}"
+      store = stored(a, {"connected" => "true"})
+      mappings = reconcile_mappings("org-#{UUID.random}", [{a, "Display"}])
+      manager = MockManager.new
+      events = StatusEvents.new(mappings, [manager] of PublisherManager)
+      events.reconcile_now(Set{a})
+
+      mappings.remove_system_modules("cs-1")
+      mappings.reconcile_systems({"cs-1"})
+      events.reconcile_now(mappings.take_reconcile)
+
+      keyed(manager.deletions).keys.should eq ["Display/1/connected"]
+    ensure
+      store.try &.clear
+    end
+
+    it "sweeps retained topics that nothing maps to" do
+      a = "mod-#{UUID.random}"
+      org = "org-#{UUID.random}"
+      store = stored(a, {"connected" => "true"})
+      mappings = reconcile_mappings(org, [{a, "Display"}])
+      events = StatusEvents.new(mappings, [MockManager.new] of PublisherManager)
+      events.reconcile_now(Set{a})
+
+      live = MqttPublisher.generate_key(mappings.published.for_module(a).first).not_nil!
+      stale = live.sub("/Display/1/", "/Display/9/")
+      publisher = SweepRecorder.new
+      events.sweep_now(publisher, Set{live, stale})
+
+      publisher.topics.should eq [stale]
+    ensure
+      store.try &.clear
+    end
+  end
+
+  private class SweepRecorder < Publisher
+    getter topics = [] of String
+
+    def publish(message : Publisher::Message)
+    end
+
+    def queue_delete_topic(topic : String) : Nil
+      topics << topic
+    end
+  end
+end
diff --git a/spec/router/control_system_router_spec.cr b/spec/router/control_system_router_spec.cr
index 4dd926b..5c98654 100644
--- a/spec/router/control_system_router_spec.cr
+++ b/spec/router/control_system_router_spec.cr
@@ -63,6 +63,30 @@ module PlaceOS::Source
       })
     end
 
+    # a Module's index is its position in the system, not in the database
+    it "numbers modules in the system's order" do
+      driver = Model::Generator.driver(module_name: "Display").save!
+      first = Model::Generator.module(driver)
+      first.custom_name = "Display"
+      first.save!
+      second = Model::Generator.module(driver)
+      second.custom_name = "Display"
+      second.save!
+
+      cs = Model::Generator.control_system
+      cs.modules = [second.id.as(String), first.id.as(String)]
+      cs.save!
+
+      mapping = Router::ControlSystem.system_modules(cs)
+      mapping[second.id.as(String)][:index].should eq 1
+      mapping[first.id.as(String)][:index].should eq 2
+
+      cs.modules = [first.id.as(String), second.id.as(String)]
+      mapping = Router::ControlSystem.system_modules(cs)
+      mapping[first.id.as(String)][:index].should eq 1
+      mapping[second.id.as(String)][:index].should eq 2
+    end
+
     it "system_zones" do
       cs = Model::Generator.control_system
       cs.id = "cs-1245"
diff --git a/spec/router/zone_router_spec.cr b/spec/router/zone_router_spec.cr
index 57cf0ae..4bd76b7 100644
--- a/spec/router/zone_router_spec.cr
+++ b/spec/router/zone_router_spec.cr
@@ -41,6 +41,24 @@ module PlaceOS::Source
       mapping.should eq({"org" => "zone-org", "building" => "zone-retagged"})
     end
 
+    # it used to be added to the hierarchy of every system, moving all their topics
+    it "leaves systems without the Zone alone when it is retagged" do
+      Model::Zone.clear
+
+      zone = Model::Generator.zone
+      zone.id = "zone-retagged"
+      zone.tags = Set{"building"}
+
+      state = Mappings::State.new
+      state.system_zones["cs-elsewhere"] = {"org" => "zone-org", "building" => "zone-other"}
+      mappings = Mappings.new(state)
+
+      Router::Zone.new(mappings, [] of PublisherManager).update_zone_mapping(zone)
+
+      mapping = mappings.read { |current| current.system_zones["cs-elsewhere"] }
+      mapping.should eq({"org" => "zone-org", "building" => "zone-other"})
+    end
+
     it "removes a Zone from system_zones when it is destroyed" do
       Model::Zone.clear
 
diff --git a/src/source/mappings.cr b/src/source/mappings.cr
index 2f1747c..ca0022f 100644
--- a/src/source/mappings.cr
+++ b/src/source/mappings.cr
@@ -2,6 +2,7 @@ require "placeos-models"
 require "rwlock"
 
 require "./constants"
+require "./published"
 
 module PlaceOS::Source
   # Consolidated data object maintaining mappings
@@ -19,6 +20,67 @@ module PlaceOS::Source
     def initialize(@state : State = State.new)
     end
 
+    # What is retained on the brokers, per Module. See `Published`
+    getter published : Published = Published.new
+
+    # Reconciliation
+    ###########################################################################
+
+    private getter reconcile_lock = Mutex.new
+    private getter pending_reconcile = Set(String).new
+
+    # Ask for what is retained for these Modules to be brought into line with
+    # the mappings as they now stand. Call it AFTER changing the mappings.
+    #
+    # Carried out by `StatusEvents`, in the same fiber that publishes status
+    # events, so the two can never interleave
+    def reconcile(module_ids : Enumerable(String)) : Nil
+      return if module_ids.empty?
+      reconcile_lock.synchronize { pending_reconcile.concat(module_ids) }
+    end
+
+    # As above, for every Module in these systems: those mapped there now, and
+    # those that had something published there, which may have left
+    def reconcile_systems(control_system_ids : Enumerable(String)) : Nil
+      ids = control_system_ids.to_set
+      return if ids.empty?
+      reconcile(
+        modules_in_systems(ids) | published.module_ids { |event| ids.includes?(event.control_system_id) }
+      )
+    end
+
+    # Takes the Modules waiting to be reconciled
+    def take_reconcile : Set(String)
+      reconcile_lock.synchronize do
+        taken = pending_reconcile.dup
+        pending_reconcile.clear
+        taken
+      end
+    end
+
+    def reconcile_pending? : Bool
+      reconcile_lock.synchronize { !pending_reconcile.empty? }
+    end
+
+    # Modules mapped into any of `control_system_ids` now
+    def modules_in_systems(control_system_ids : Enumerable(String)) : Set(String)
+      ids = control_system_ids.to_set
+      read do |state|
+        state.system_modules.each_with_object(Set(String).new) do |(module_id, mappings), found|
+          found << module_id if mappings.any? { |mapping| ids.includes?(mapping[:control_system_id]) }
+        end
+      end
+    end
+
+    # Systems whose hierarchy includes `zone_id` now
+    def systems_in_zone(zone_id : String) : Set(String)
+      read do |state|
+        state.system_zones.each_with_object(Set(String).new) do |(control_system_id, zones), found|
+          found << control_system_id if zones.values.includes?(zone_id)
+        end
+      end
+    end
+
     # State Keys
     ###########################################################################
 
@@ -67,18 +129,6 @@ module PlaceOS::Source
       end
     end
 
-    # Every status topic currently mapped for a Module.
-    #
-    # Called *before* a mapping is rewritten, so the caller gets the topics as
-    # they exist right now — the ones that have to be removed once the Module
-    # moves or goes away. The statuses themselves come from the Module's
-    # storage, the same source a state resync reads
-    def current_status_events(module_id : String, statuses : Enumerable(String)) : Array(Status)
-      statuses.flat_map do |status|
-        status_events?(module_id, status) || [] of Status
-      end
-    end
-
     record Status,
       status : String,
       index : Int32,
@@ -176,6 +226,18 @@ module PlaceOS::Source
       end
     end
 
+    # The module mappings of one system, as they stand
+    #
+    # NOTE:: call with the mappings lock held or not; it takes a read lock of
+    # its own, and `RWLock` allows nested reads
+    def system_modules_for(control_system_id : String) : Hash(String, SystemModule)
+      read do |state|
+        state.system_modules.each_with_object({} of String => SystemModule) do |(module_id, list), found|
+          list.each { |mapping| found[module_id] = mapping if mapping[:control_system_id] == control_system_id }
+        end
+      end
+    end
+
     # Remove System Module Mappings
     def remove_system_modules(control_system_id : String)
       # Clear mappings of all references to control_system_id
diff --git a/src/source/published.cr b/src/source/published.cr
new file mode 100644
index 0000000..c16825d
--- /dev/null
+++ b/src/source/published.cr
@@ -0,0 +1,52 @@
+require "./mappings"
+
+module PlaceOS::Source
+  # What is retained on the brokers right now, per Module.
+  #
+  # Recorded as each status is published or removed, so it is known what to
+  # take away when a topic stops applying. It does not depend on the Module's
+  # stored state or on the mappings, both of which may already have changed by
+  # the time a change reaches us: a destroyed Module's storage is cleared by
+  # core, and its system may drop it from the mappings before the Module's own
+  # delete event arrives. Working the old topics out from either of those was
+  # best effort, and lost them.
+  class Published
+    private getter lock = Mutex.new
+
+    # module_id => statuses published for it, one per system it is mapped in
+    private getter events = Hash(String, Set(Mappings::Status)).new
+
+    def add(event : Mappings::Status) : Nil
+      lock.synchronize do
+        (events[event.module_id] ||= Set(Mappings::Status).new) << event
+      end
+    end
+
+    def remove(event : Mappings::Status) : Nil
+      lock.synchronize do
+        if set = events[event.module_id]?
+          set.delete(event)
+          events.delete(event.module_id) if set.empty?
+        end
+      end
+    end
+
+    def for_module(module_id : String) : Set(Mappings::Status)
+      lock.synchronize { events[module_id]?.try(&.dup) || Set(Mappings::Status).new }
+    end
+
+    # Modules with anything published that matches the block
+    def module_ids(& : Mappings::Status -> Bool) : Set(String)
+      lock.synchronize do
+        events.each_with_object(Set(String).new) do |(module_id, set), ids|
+          ids << module_id if set.any? { |event| yield event }
+        end
+      end
+    end
+
+    # Every event published, for comparing with what a broker holds
+    def all : Array(Mappings::Status)
+      lock.synchronize { events.values.flat_map(&.to_a) }
+    end
+  end
+end
diff --git a/src/source/publishing/mqtt_broker_manager.cr b/src/source/publishing/mqtt_broker_manager.cr
index f147053..f672cb9 100644
--- a/src/source/publishing/mqtt_broker_manager.cr
+++ b/src/source/publishing/mqtt_broker_manager.cr
@@ -36,6 +36,11 @@ module PlaceOS::Source
       end
     end
 
+    # The publishers as they are now, one per Broker
+    def publishers : Array(MqttPublisher)
+      read_publishers(&.values.dup)
+    end
+
     def deletes_pending? : Bool
       read_publishers do |publishers|
         publishers.values.any?(&.deletes_pending?)
@@ -72,12 +77,17 @@ module PlaceOS::Source
 
     # Create a `MqttPublisher` for the `Broker`
     #
-    protected def create_publisher(broker : Model::Broker) : Resource::Result
+    protected def create_publisher(broker : Model::Broker, retry : Bool = true) : Resource::Result
       broker_id = broker.id.as(String)
       begin
         publisher = MqttPublisher.new(broker)
-      rescue e : IO::Error
-        Log.error(exception: e) { "failed to connect to broker on #{broker.host}:#{broker.port}" }
+      rescue e
+        # 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
+        # that starts after source, or restarts under it, is ordinary
+        Log.error(exception: e) { "failed to connect to broker on #{broker.host}:#{broker.port}#{", retrying" if retry}" }
+        retry_publisher(broker_id) if retry
         return Resource::Result::Error
       end
 
@@ -89,6 +99,11 @@ module PlaceOS::Source
         !existing.nil?
       end
 
+      publisher.on_reconnect = -> {
+        Log.info { "reconnected to Broker<#{broker_id}>, republishing state" }
+        on_broker_ready.try &.call(broker_id)
+        nil
+      }
       publisher.start
 
       # Trigger state sync callback if this is a new broker after startup.
@@ -106,6 +121,30 @@ module PlaceOS::Source
       Resource::Result::Success
     end
 
+    private getter retrying = Set(String).new
+    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
+    protected def retry_publisher(broker_id : String) : Nil
+      return unless retrying_lock.synchronize { retrying.add?(broker_id) }
+
+      spawn do
+        delay = 1.second
+        begin
+          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?
+          end
+        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
diff --git a/src/source/publishing/mqtt_publisher.cr b/src/source/publishing/mqtt_publisher.cr
index 247f170..87d4d21 100644
--- a/src/source/publishing/mqtt_publisher.cr
+++ b/src/source/publishing/mqtt_publisher.cr
@@ -36,8 +36,31 @@ module PlaceOS::Source
     # and falling back to 3.1.1 when the broker will not take it
     getter client : ::MQTT::Client
 
+    # Called once the client has re-established a dropped connection
+    property on_reconnect : Proc(Nil)? = nil
+
+    @connections = Atomic(Int32).new(0)
+
     def initialize(@broker : PlaceOS::Model::Broker)
-      @client = MqttPublisher.client(@broker)
+      @client = MqttPublisher.client(@broker) { connection_opened }
+    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
     end
 
     # The protocol version that was actually negotiated with this broker
@@ -67,7 +90,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) : ::MQTT::Client
+    def self.client(broker : PlaceOS::Model::Broker, &on_transport : -> Nil) : ::MQTT::Client
       Log.debug { {message: "creating MQTT client", host: broker.host, port: broker.port.to_s, tls: broker.tls} }
 
       host = broker.host
@@ -78,6 +101,7 @@ 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
@@ -100,6 +124,13 @@ module PlaceOS::Source
     end
 
     protected def publish(message : Message)
+      # A status the driver cleared, or a model that was destroyed, is gone:
+      # remove it rather than retain a `null` in its place. The store has no
+      # value for it, so a retained `{"value": null}` would disagree with it
+      if MqttPublisher.removal?(message.payload)
+        return delete(message)
+      end
+
       if key = MqttPublisher.generate_key(message.data)
         # Sanitize the message payload according the Broker's filters
         payload = broker_lock.read do
@@ -138,7 +169,16 @@ module PlaceOS::Source
     def delete(message : Message) : Nil
       key = MqttPublisher.generate_key(message.data)
       return unless key
+      delete_topic(key)
+    end
+
+    # Whether a payload means the value no longer exists. Redis storage
+    # publishes `"null"` when a status is deleted or the store is cleared
+    def self.removal?(payload : String?) : Bool
+      payload.nil? || payload == "null"
+    end
 
+    def delete_topic(key : String) : Nil
       Log.trace { {message: "removing retained value", key: key} }
 
       Retriable.retry(
@@ -153,7 +193,42 @@ module PlaceOS::Source
         client.publish(topic: key, payload: "", retain: true)
       end
     rescue e
-      Log.error(exception: e) { "error removing retained value for #{message.data}" }
+      Log.error(exception: e) { "error removing retained value for #{key}" }
+    end
+
+    # Every status topic the broker currently retains, as a client connecting
+    # now would be handed 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) : Set(String)
+      filter = File.join(MQTT_NAMESPACE, "+", "state", "#")
+      topics = Set(String).new
+      lock = Mutex.new
+      last = Time.instant
+
+      client.subscribe(filter) do |topic, payload, retained|
+        next unless retained && !payload.empty?
+        lock.synchronize do
+          topics << topic
+          last = Time.instant
+        end
+      end
+
+      begin
+        started = last = Time.instant
+        loop do
+          sleep 100.milliseconds
+          now = Time.instant
+          break if lock.synchronize { now - last >= quiet } || now - started >= limit
+        end
+      ensure
+        client.unsubscribe(filter) rescue nil
+      end
+
+      lock.synchronize { topics.dup }
     end
 
     # Key generation
diff --git a/src/source/publishing/publisher.cr b/src/source/publishing/publisher.cr
index e03d782..3c90b5c 100644
--- a/src/source/publishing/publisher.cr
+++ b/src/source/publishing/publisher.cr
@@ -11,16 +11,24 @@ module PlaceOS::Source
       timestamp : Time
     )
 
-    getter message_queue : Channel(Message) = Channel(Message).new(StatusEvents::BATCH_SIZE)
+    # A retained value to remove: either the key a message would publish to, or
+    # a topic by name, for one that nothing maps to any more
+    record(
+      Deletion,
+      message : Message? = nil,
+      topic : String? = nil
+    )
 
-    # Keys whose retained value has to be removed, because the topic they were
-    # published under no longer exists — a renamed or reindexed Module, or one
-    # that was destroyed.
+    # Publications and deletions, in the order they were made.
     #
-    # NOTE:: this is drained ahead of `message_queue`. A rename removes the old
-    # topic and republishes under the new one, and if the publish were to win
-    # the race the delete would take the new value straight back out again
-    getter delete_queue : Channel(Message) = Channel(Message).new(StatusEvents::BATCH_SIZE)
+    # NOTE:: ONE queue, so a deletion can never overtake a publication made
+    # before it, or be overtaken by one made after. Deletions used to have a
+    # queue of their own that was drained first, which let a removal jump ahead
+    # of a stale value still waiting to be published — the value then came
+    # back. Ordering is enough now because a deletion is only ever queued for a
+    # topic that nothing maps to (`Reconciler`), so a later publication to the
+    # same topic is, by construction, a value that belongs there
+    getter message_queue : Channel(Message | Deletion) = Channel(Message | Deletion).new(StatusEvents::BATCH_SIZE)
 
     getter processed : UInt64 = 0_u64
     getter deleted : UInt64 = 0_u64
@@ -32,8 +40,17 @@ module PlaceOS::Source
 
     # Queues removal of whatever is retained for the message's key
     def queue_delete(message : Message) : Nil
+      queue_deletion(Deletion.new(message: message))
+    end
+
+    # Queues removal of whatever is retained under `topic`
+    def queue_delete_topic(topic : String) : Nil
+      queue_deletion(Deletion.new(topic: topic))
+    end
+
+    private def queue_deletion(deletion : Deletion) : Nil
       @pending_deletes.add(1)
-      delete_queue.send(message)
+      message_queue.send(deletion)
     end
 
     abstract def publish(message : Message)
@@ -44,6 +61,10 @@ module PlaceOS::Source
     def delete(message : Message) : Nil
     end
 
+    # Removes whatever is retained under `topic`. Nothing by default, as above
+    def delete_topic(topic : String) : Nil
+    end
+
     def commit : Nil
     end
 
@@ -59,27 +80,23 @@ module PlaceOS::Source
 
     def stop
       message_queue.close
-      delete_queue.close
     end
 
     private def consume_messages
       while !message_queue.closed?
-        # NOTE:: `select` makes a non-blocking pass over its branches in order,
-        # so listing deletions first is what gives them priority when both
-        # queues are ready. A rename removes the old topic and republishes
-        # under the new one, and a deletion arriving late would take the new
-        # value straight back out
         select
-        when deletion = delete_queue.receive?
-          handle_deletion(deletion) if deletion
-        when message = message_queue.receive?
-          if message
+        when item = message_queue.receive?
+          case item
+          in Message
             begin
-              publish(message)
+              publish(item)
               @processed += 1_u64
             rescue error
-              Log.warn(exception: error) { "publishing message: #{message}" }
+              Log.warn(exception: error) { "publishing message: #{item}" }
             end
+          in Deletion
+            handle_deletion(item)
+          in Nil
           end
         when timeout(10.seconds)
           # commit any buffered messages that have not been published yet
@@ -88,11 +105,15 @@ module PlaceOS::Source
       end
     end
 
-    private def handle_deletion(message : Message) : Nil
-      delete(message)
+    private def handle_deletion(deletion : Deletion) : Nil
+      if message = deletion.message
+        delete(message)
+      elsif topic = deletion.topic
+        delete_topic(topic)
+      end
       @deleted += 1_u64
     rescue error
-      Log.warn(exception: error) { "deleting retained message: #{message}" }
+      Log.warn(exception: error) { "deleting retained message: #{deletion}" }
     ensure
       @pending_deletes.sub(1)
     end
diff --git a/src/source/router/control_system_router.cr b/src/source/router/control_system_router.cr
index 52ccae4..77d0cfa 100644
--- a/src/source/router/control_system_router.cr
+++ b/src/source/router/control_system_router.cr
@@ -25,34 +25,57 @@ module PlaceOS::Source::Router
       super()
     end
 
-    # - Create a system_zones mapping
-    # - Create a system_modules mapping
-    def handle_create(control_system : PlaceOS::Model::ControlSystem) : Resource::Result
-      zone_mappings(control_system)
-      module_mappings(control_system)
+    # Events are processed concurrently and in no particular order: pg-orm and
+    # `Resource` each handle every change on a fiber of its own. So two quick
+    # edits to one system can be handled newest first, and an event's own copy
+    # of the system, and its `*_changed?` flags, cannot be trusted to be the
+    # latest. Each event instead takes this lock, reads the system as it is
+    # now, and rebuilds its mappings from that: whichever runs last leaves
+    # them matching the database
+    private getter sync_lock = Mutex.new
 
-      Resource::Result::Success
+    def handle_create(control_system : PlaceOS::Model::ControlSystem) : Resource::Result
+      sync(control_system.id.as(String))
     end
 
-    # - Create a system_zones mapping if zones have changed
-    # - Update system_modules mapping if modules have changed
     def handle_update(control_system : PlaceOS::Model::ControlSystem) : Resource::Result
-      zone_mappings(control_system) if control_system.zones_changed?
-      module_mappings(control_system) if control_system.modules_changed?
-
-      Resource::Result::Success
+      sync(control_system.id.as(String))
     end
 
-    # - Remove system_modules references to ControlSystem
-    # - Remove system_zones for ControlSystem
     def handle_delete(control_system : PlaceOS::Model::ControlSystem) : Resource::Result
-      control_system_id = control_system.id.as(String)
+      sync(control_system.id.as(String))
+    end
 
-      mappings.write do |state|
-        state.system_zones.delete(control_system_id)
-      end
+    # Brings the mappings for a system into line with the database, and has
+    # what is retained for it reconciled if they changed.
+    #
+    # Zones are topic segments, and a module's index is its position among the
+    # system's modules of the same name, so a change to either moves topics.
+    # Reconciling removes the old ones and publishes current values at the new
+    # ones — which also covers a module whose state was written before the
+    # system mapped it, and so was published nowhere
+    def sync(control_system_id : String) : Resource::Result
+      sync_lock.synchronize do
+        current = PlaceOS::Model::ControlSystem.find?(control_system_id)
+
+        zones = current.try { |cs| Router::ControlSystem.system_zones(cs) }
+        modules = current.try { |cs| Router::ControlSystem.system_modules(cs) }
+
+        changed = mappings.read do |state|
+          state.system_zones[control_system_id]? != zones ||
+            mappings.system_modules_for(control_system_id) != (modules || {} of String => Mappings::SystemModule)
+        end
 
-      mappings.remove_system_modules(control_system_id)
+        if zones && modules
+          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
+
+        mappings.reconcile_systems({control_system_id}) if changed
+      end
 
       Resource::Result::Success
     end
@@ -62,7 +85,10 @@ module PlaceOS::Source::Router
 
       hierarchy_zones = Mappings.hierarchy_zones(control_system)
 
-      return Resource::Result::Skipped if hierarchy_zones.empty? && !action.deleted?
+      # NOTE:: an update is processed even when the system is in no scoped zone
+      # any more: it may have just been moved out of one, and what it published
+      # while it was in scope has to go
+      return Resource::Result::Skipped if hierarchy_zones.empty? && action.created?
 
       hierarchy_zones.each { |zone| publish_metadata(zone, control_system) }
 
@@ -76,26 +102,6 @@ module PlaceOS::Source::Router
       end
     end
 
-    # Create/update a mapping for ControlSystem's Zones
-    def zone_mappings(control_system : PlaceOS::Model::ControlSystem)
-      system_zone_mappings = Router::ControlSystem.system_zones(control_system)
-
-      mappings.write do |state|
-        state.system_zones[control_system.id.as(String)] = system_zone_mappings
-      end
-
-      system_zone_mappings
-    end
-
-    # Create/update mappings for ControlSystem's Modules
-    def module_mappings(control_system : PlaceOS::Model::ControlSystem)
-      system_module_mappings = Router::ControlSystem.system_modules(control_system)
-
-      mappings.set_system_modules(control_system.id.as(String), system_module_mappings)
-
-      system_module_mappings
-    end
-
     def self.system_zones(control_system : PlaceOS::Model::ControlSystem, zones : Array(PlaceOS::Model::Zone)? = nil) : Hash(String, String)
       # Generate zone mappings
       # NOTE: Duplication decessary due to compiler bug with iteration
@@ -149,10 +155,17 @@ module PlaceOS::Source::Router
       PlaceOS::Model::Zone.find_all(zone_list)
     end
 
-    # Get modules
+    # Get modules, in the system's order.
+    #
+    # NOTE:: the order is load-bearing: a Module's index is its position among
+    # the system's Modules of the same name. `find_all` returns rows in
+    # whatever order the database does, so indexes used to follow the Module
+    # ids rather than the system, and reordering a system changed nothing that
+    # was published
     protected def self._system_modules(control_system : PlaceOS::Model::ControlSystem)
       module_ids = control_system.modules || [] of String
-      PlaceOS::Model::Module.find_all(module_ids)
+      found = PlaceOS::Model::Module.find_all(module_ids).to_h { |mod| {mod.id.as(String), mod} }
+      module_ids.compact_map { |id| found[id]? }
     end
   end
 end
diff --git a/src/source/router/driver_router.cr b/src/source/router/driver_router.cr
index f7f52bb..b5891d7 100644
--- a/src/source/router/driver_router.cr
+++ b/src/source/router/driver_router.cr
@@ -2,7 +2,6 @@ require "placeos-models/driver"
 require "placeos-resource"
 
 require "../mappings"
-require "./module_router"
 require "../publishing/publish_metadata"
 require "../publishing/publisher_manager"
 
@@ -40,13 +39,11 @@ module PlaceOS::Source::Router
           state.drivers.compact_map { |mod_id, id| mod_id if id == driver_id }
         end
 
-        module_router = Router::Module.new(mappings, publisher_managers)
-        module_ids.each { |mod_id| module_router.remove_retained_state(mod_id) }
-
         mappings.write do |state|
           # Remove references to this Driver
           state.drivers.reject! { |_, id| id == driver_id }
         end
+        mappings.reconcile(module_ids)
       end
 
       Resource::Result::Success
diff --git a/src/source/router/module_router.cr b/src/source/router/module_router.cr
index e4e81fa..53aa8b5 100644
--- a/src/source/router/module_router.cr
+++ b/src/source/router/module_router.cr
@@ -1,4 +1,3 @@
-require "placeos-driver/storage"
 require "placeos-models/module"
 require "placeos-resource"
 
@@ -18,36 +17,6 @@ module PlaceOS::Source::Router
       super()
     end
 
-    # Queue removal of everything this Module currently publishes.
-    #
-    # Best effort: the statuses come from the Module's storage, which another
-    # service may already have cleared by the time a delete reaches us. What is
-    # still there gets removed, and anything already gone was never ours to find
-    def remove_retained_state(module_id : String) : Nil
-      return if publisher_managers.empty?
-
-      statuses = begin
-        PlaceOS::Driver::RedisStorage.new(module_id).keys
-      rescue error
-        Log.debug(exception: error) { "no stored state for Module<#{module_id}>" }
-        [] of String
-      end
-      return if statuses.empty?
-
-      # resolved against the mapping as it stands, so these are the old topics
-      events = mappings.current_status_events(module_id, statuses)
-      return if events.empty?
-
-      # the payload is discarded — a removal is a zero length retained publish
-      timestamp = Time.utc
-      events.each do |event|
-        message = Publisher::Message.new(event, nil, timestamp)
-        publisher_managers.each(&.broadcast_delete(message))
-      end
-
-      Log.info { "queued removal of #{events.size} retained value(s) for Module<#{module_id}>" }
-    end
-
     def handle_create(mod : Model::Module)
       mappings.write do |state|
         state.drivers[mod.id.as(String)] = mod.driver_id.as(String)
@@ -59,18 +28,14 @@ module PlaceOS::Source::Router
     def handle_update(mod : Model::Module)
       module_id = mod.id.as(String)
 
-      # Update all `system_mappings` if Module's `custom_name` changed.
-      #
       # NOTE:: the name is a topic segment, so a rename moves every status this
-      # Module publishes. The old topics have to be removed *before* the mapping
-      # is rewritten, while they can still be resolved
+      # Module publishes, and can shift the index of others with either name
       if mod.custom_name_changed?
-        remove_retained_state(module_id)
-
-        # Update the `system_module` entry for each ControlSystem that has a reference to the Module
-        Model::ControlSystem.by_module_id(module_id).each do |cs|
+        system_ids = Model::ControlSystem.by_module_id(module_id).map do |cs|
           mappings.set_system_modules(cs.id.as(String), Router::ControlSystem.system_modules(cs))
+          cs.id.as(String)
         end
+        mappings.reconcile_systems(system_ids)
       end
 
       Resource::Result::Success
@@ -79,8 +44,10 @@ module PlaceOS::Source::Router
     def handle_delete(mod : Model::Module)
       module_id = mod.id.as(String)
 
-      # again, before the mapping goes, or the topics cannot be resolved
-      remove_retained_state(module_id)
+      # the systems it was in, while the mapping still says so
+      system_ids = mappings.read do |state|
+        state.system_modules[module_id]?.try(&.map(&.[:control_system_id])) || [] of String
+      end
 
       mappings.write do |state|
         # Remove reference in drivers
@@ -88,6 +55,11 @@ module PlaceOS::Source::Router
         state.system_modules.delete(module_id)
       end
 
+      # What it published is known from the registry, not from its storage,
+      # which core may already have cleared
+      mappings.reconcile({module_id})
+      mappings.reconcile_systems(system_ids)
+
       Resource::Result::Success
     end
 
diff --git a/src/source/router/zone_router.cr b/src/source/router/zone_router.cr
index 8286431..bd0ffa6 100644
--- a/src/source/router/zone_router.cr
+++ b/src/source/router/zone_router.cr
@@ -5,6 +5,7 @@ require "../constants"
 require "../mappings"
 require "../publishing/publish_metadata"
 require "../publishing/publisher_manager"
+require "./control_system_router"
 
 module PlaceOS::Source::Router
   # Zone router (if scoped)...
@@ -48,22 +49,42 @@ module PlaceOS::Source::Router
       end
     end
 
-    # Handle Updates to existing Zone tags
+    # Handle a change to a Zone's tags, or its removal.
     #
+    # Only systems that have the zone are touched. Every system's mapping used
+    # to be passed through `Mappings.zone_mapping`, which adds the zone to any
+    # mapping that does not have it yet — so retagging one zone put it in the
+    # hierarchy of every system, and moved all of their topics.
+    #
+    # Mappings that have the zone are patched; then every system the database
+    # says is in the zone is rebuilt from it, which also covers a zone that has
+    # just gained or lost its hierarchy tag
     def update_zone_mapping(zone : PlaceOS::Model::Zone)
-      zone_tag = Mappings.hierarchy_tag?(zone)
-      # Ignore zone if tags do not fall in the hierarchy
-      return Resource::Result::Skipped if zone_tag.nil?
-
       zone_id = zone.id.as(String)
-      destroyed = zone.destroyed?
+      zone_tag = Mappings.hierarchy_tag?(zone)
+      remove = zone.destroyed? || zone_tag.nil?
 
+      mapped = mappings.systems_in_zone(zone_id)
       mappings.write do |state|
-        state.system_zones.transform_values! do |mapping|
-          Mappings.zone_mapping(zone_id, zone_tag, mapping, destroyed)
+        mapped.each do |control_system_id|
+          next unless mapping = state.system_zones[control_system_id]?
+          if remove
+            mapping.reject! { |_, id| id == zone_id }
+          else
+            Mappings.zone_mapping(zone_id, zone_tag.as(String), mapping)
+          end
         end
       end
 
+      in_zone = remove ? [] of PlaceOS::Model::ControlSystem : PlaceOS::Model::ControlSystem.by_zone_id(zone_id).to_a
+      rebuilt = in_zone.map { |cs| {cs.id.as(String), Router::ControlSystem.system_zones(cs)} }
+      mappings.write do |state|
+        rebuilt.each { |(control_system_id, mapping)| state.system_zones[control_system_id] = mapping }
+      end
+
+      # the zone is a topic segment of every status in those systems
+      mappings.reconcile_systems(mapped | rebuilt.map(&.[0]).to_set)
+
       Resource::Result::Success
     end
   end
diff --git a/src/source/status_events.cr b/src/source/status_events.cr
index 111c3ee..7cfe69a 100644
--- a/src/source/status_events.cr
+++ b/src/source/status_events.cr
@@ -6,6 +6,7 @@ require "placeos-driver/storage"
 require "./mappings"
 require "./publishing/publisher"
 require "./publishing/publisher_manager"
+require "./publishing/mqtt_broker_manager"
 
 module PlaceOS::Source
   class StatusEvents
@@ -131,6 +132,7 @@ module PlaceOS::Source
         values:  status_updated.to_s,
       } }
       self.initial_sync_complete = true
+      spawn { sweep_brokers }
     end
 
     # Trigger a state resync - useful when new brokers are added
@@ -173,6 +175,8 @@ module PlaceOS::Source
         modules: mods_mapped.to_s,
         values:  status_updated.to_s,
       } }
+
+      sweep_brokers
     end
 
     protected def handle_pevent(pattern : String, channel : String, payload : String)
@@ -200,6 +204,13 @@ module PlaceOS::Source
     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)
+          end
+
           batch = build_batch
           if batch.empty?
             sleep PROCESSING_INTERVAL
@@ -252,6 +263,13 @@ module PlaceOS::Source
       } }
 
       events.try &.each do |event|
+        # what is retained follows the value: a cleared status is removed
+        if MqttPublisher.removal?(payload)
+          mappings.published.remove(event)
+        else
+          mappings.published.add(event)
+        end
+
         message = Publisher::Message.new(event, payload, timestamp)
         publisher_managers.each do |manager|
           Log.trace { "broadcasting message to #{manager.class}" }
@@ -260,6 +278,133 @@ module PlaceOS::Source
       end
     end
 
+    # Reconciliation
+    ###########################################################################
+
+    # Brings what is retained for each Module into line with the mappings as
+    # they stand: topics that no longer apply are removed, and topics that now
+    # apply but were never published get the Module's current value.
+    #
+    # Runs in the event processing fiber, so the registry, the mappings it is
+    # compared with and the values read here cannot be overtaken by a status
+    # event: one that arrives meanwhile is processed afterwards, against the
+    # new mappings, with a value at least as new as the one read here.
+    #
+    # Public for specs; otherwise only the processing fiber calls it
+    def reconcile_now(module_ids : Set(String)) : Nil
+      return if module_ids.empty?
+
+      removals = [] of Mappings::Status
+      additions = [] of {Mappings::Status, String}
+      # topics something maps to now, which a removal must leave alone: in a
+      # reorder, Display_1 stops being one Module's and becomes another's
+      valid = Set(String).new
+
+      module_ids.each do |module_id|
+        before = mappings.published.for_module(module_id)
+        values = begin
+          PlaceOS::Driver::RedisStorage.new(module_id).to_h
+        rescue error
+          Log.warn(exception: error) { "reading state of Module<#{module_id}> to reconcile" }
+          next
+        end
+
+        after = Set(Mappings::Status).new
+        values.each_key do |status|
+          mappings.status_events?(module_id, status).try &.each { |event| after << event }
+        end
+
+        after.each do |event|
+          if key = MqttPublisher.generate_key(event)
+            valid << key
+          end
+        end
+        (before - after).each { |event| removals << event }
+        (after - before).each { |event| additions << {event, values[event.status]} }
+      end
+
+      timestamp = Time.utc
+      removals.each do |event|
+        mappings.published.remove(event)
+        key = MqttPublisher.generate_key(event)
+        next if key.nil? || valid.includes?(key)
+        message = Publisher::Message.new(event, nil, timestamp)
+        publisher_managers.each(&.broadcast_delete(message))
+      end
+
+      additions.each do |(event, payload)|
+        mappings.published.add(event)
+        message = Publisher::Message.new(event, payload, timestamp)
+        publisher_managers.each(&.broadcast(message))
+      end
+
+      unless removals.empty? && additions.empty?
+        Log.info { {
+          message: "reconciled retained state",
+          modules: module_ids.size.to_s,
+          removed: removals.size.to_s,
+          added:   additions.size.to_s,
+        } }
+      end
+    end
+
+    # Broker sweep
+    ###########################################################################
+
+    private getter sweep_lock = Mutex.new
+    @pending_sweep : {MqttPublisher, Set(String)}? = nil
+
+    # Removes every status topic a broker retains that nothing maps to.
+    #
+    # The registry of what is published starts empty, so whatever a broker
+    # held from before this process started is unaccounted for: Modules
+    # deleted or moved while source was down, or by a version of source that
+    # did not clean up after them. Once the initial sync has been processed
+    # the registry is complete, and anything else the broker holds is stale.
+    def sweep_brokers : Nil
+      # the registry is only complete once every synced event is processed
+      until stopped? || (synchronize { event_container.empty? } && !mappings.reconcile_pending?)
+        sleep 100.milliseconds
+      end
+
+      publisher_managers.each do |manager|
+        next unless manager.is_a?(MqttBrokerManager)
+        manager.publishers.each do |publisher|
+          begin
+            topics = publisher.retained_state_topics
+            sweep_lock.synchronize { @pending_sweep = {publisher, topics} }
+            # one at a time, so the processing fiber takes each in turn
+            while sweep_lock.synchronize { !@pending_sweep.nil? } && !stopped?
+              sleep 50.milliseconds
+            end
+          rescue error
+            Log.warn(exception: error) { "unable to read retained topics from Broker<#{publisher.broker.id}>, stale topics remain" }
+          end
+        end
+      end
+    end
+
+    private def take_sweep : {MqttPublisher, Set(String)}?
+      sweep_lock.synchronize do
+        sweep = @pending_sweep
+        @pending_sweep = nil
+        sweep
+      end
+    end
+
+    # Public for specs; otherwise only the processing fiber calls it
+    def sweep_now(publisher : Publisher, retained : Set(String), label : String = "") : Nil
+      valid = mappings.published.all.compact_map { |event| MqttPublisher.generate_key(event) }.to_set
+      stale = retained - valid
+      stale.each { |topic| publisher.queue_delete_topic(topic) }
+      Log.info { {
+        message:  "swept retained state",
+        broker:   label,
+        retained: retained.size.to_s,
+        removed:  stale.size.to_s,
+      } }
+    end
+
     def self.parse_channel(channel : String) : {String, String}
       _, module_id, status = channel.split('/')
       {module_id, status}