From f69196fda6141f6f6a851c3a9b4d50596abb03a5 Mon Sep 17 00:00:00 2001
From: Stephen von Takach <steve@place.technology>
Date: Thu, 15 Oct 2026 00:28:13 +1100
Subject: [PATCH] fix(mqtt): keep retained state topics accurate

Brokers held retained state that no longer applied, and missed state that
did. Clients that trust the broker (stagehand's hot list) alerted on deleted
modules and on a module's old index after a reorder.

- number modules by the system's own module order, not database order
- record what is published per module, and after any system, module,
  driver or zone change remove topics that no longer apply and publish
  current values at new ones
- handle changes to one system in order, re-reading it from the database
- publish state written before the module's system was known
- only touch the systems a zone change applies to
- send publishes and deletions through one ordered queue
- remove a cleared status from the broker rather than retaining null
- republish everything when a broker reconnects, and retry a broker that
  is down at startup with backoff
- sweep retained state topics nothing maps to after the initial sync

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 spec/publishing/publisher_spec.cr            |  32 ++--
 spec/publishing/rename_spec.cr               |  16 +-
 spec/reconcile_spec.cr                       | 162 +++++++++++++++++++
 spec/router/control_system_router_spec.cr    |  24 +++
 spec/router/zone_router_spec.cr              |  18 +++
 src/source/mappings.cr                       |  86 ++++++++--
 src/source/published.cr                      |  52 ++++++
 src/source/publishing/mqtt_broker_manager.cr |  45 +++++-
 src/source/publishing/mqtt_publisher.cr      |  81 +++++++++-
 src/source/publishing/publisher.cr           |  69 +++++---
 src/source/router/control_system_router.cr   |  97 ++++++-----
 src/source/router/driver_router.cr           |   5 +-
 src/source/router/module_router.cr           |  54 ++-----
 src/source/router/zone_router.cr             |  37 ++++-
 src/source/status_events.cr                  | 145 +++++++++++++++++
 15 files changed, 763 insertions(+), 160 deletions(-)
 create mode 100644 spec/reconcile_spec.cr
 create mode 100644 src/source/published.cr

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}