From 1c97e8fddbb0be6de07f957bf2c9fee91b64bd66 Mon Sep 17 00:00:00 2001
From: Stephen von Takach <steve@place.technology>
Date: Thu, 15 Oct 2026 00:53:23 +1100
Subject: [PATCH 1/3] fix(mqtt): keep startup memory and time flat on large
 estates

#65 made startup reconcile every Module in the estate in one batch: each
system mapped as the routers loaded queued a reconcile, so the first pass
read every Module's state into memory at once, on top of the initial sync
that already publishes it with backpressure. Mapping each system also
scanned every Module, so loading was quadratic in the estate's size.

- systems read by the initial load are mapped from that row and queue no
  reconcile; the initial sync covers them
- index system modules by system, and the published registry by system,
  so per-system work no longer scans everything
- reconcile reads status names up front and values a Module at a time
- the broker sweep checks each retained topic as it arrives and keeps only
  the stale ones, rather than comparing two sets of every topic

Mapping 20,000 systems of 5 modules: 9.8s before, 0.04s after.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 spec/mappings_spec.cr                      | 38 +++++++++
 spec/published_spec.cr                     | 42 ++++++++++
 spec/publishing/rename_spec.cr             |  2 +-
 spec/router/control_system_router_spec.cr  | 26 ++++++
 spec/router/module_router_spec.cr          | 12 +--
 spec/spec_helper.cr                        |  2 +-
 src/source/mappings.cr                     | 63 +++++++++++----
 src/source/published.cr                    | 41 +++++++---
 src/source/publishing/mqtt_publisher.cr    | 19 ++++-
 src/source/router/control_system_router.cr | 20 +++--
 src/source/router/module_router.cr         |  2 +-
 src/source/status_events.cr                | 92 +++++++++++++++++-----
 12 files changed, 293 insertions(+), 66 deletions(-)
 create mode 100644 spec/published_spec.cr

diff --git a/spec/mappings_spec.cr b/spec/mappings_spec.cr
index 175e593..eaa68fe 100644
--- a/spec/mappings_spec.cr
+++ b/spec/mappings_spec.cr
@@ -33,6 +33,44 @@ module PlaceOS::Source
       end
     end
 
+    describe "system modules" do
+      it "keeps each system's modules apart" do
+        mappings = Mappings.new
+        mappings.set_system_modules("cs-1", {
+          "mod-a" => {name: "Display", control_system_id: "cs-1", index: 1},
+          "mod-b" => {name: "Display", control_system_id: "cs-1", index: 2},
+        })
+        mappings.set_system_modules("cs-2", {
+          "mod-a" => {name: "Screen", control_system_id: "cs-2", index: 1},
+        })
+
+        mappings.modules_in_systems({"cs-1"}).should eq Set{"mod-a", "mod-b"}
+        mappings.system_modules_for("cs-2").should eq({"mod-a" => {name: "Screen", control_system_id: "cs-2", index: 1}})
+
+        # replacing a system's modules leaves the other system's alone
+        mappings.set_system_modules("cs-1", {
+          "mod-b" => {name: "Display", control_system_id: "cs-1", index: 1},
+        })
+        mappings.modules_in_systems({"cs-1"}).should eq Set{"mod-b"}
+        mappings.read(&.system_modules["mod-a"]).should eq [{name: "Screen", control_system_id: "cs-2", index: 1}]
+
+        mappings.remove_system_modules("cs-2")
+        mappings.read(&.system_modules.has_key?("mod-a")).should be_false
+        mappings.system_modules_for("cs-2").should be_empty
+      end
+
+      it "forgets a deleted module in every system" do
+        mappings = Mappings.new
+        mappings.set_system_modules("cs-1", {"mod-a" => {name: "Display", control_system_id: "cs-1", index: 1}})
+        mappings.set_system_modules("cs-2", {"mod-a" => {name: "Display", control_system_id: "cs-2", index: 1}})
+
+        mappings.write(&.unmap_module("mod-a"))
+
+        mappings.modules_in_systems({"cs-1", "cs-2"}).should be_empty
+        mappings.read(&.system_module_ids).should be_empty
+      end
+    end
+
     describe "hierarchy_zones" do
       it "gets scope zones for ControlSystem" do
         Model::ControlSystem.clear
diff --git a/spec/published_spec.cr b/spec/published_spec.cr
new file mode 100644
index 0000000..6b3bdcf
--- /dev/null
+++ b/spec/published_spec.cr
@@ -0,0 +1,42 @@
+require "./spec_helper"
+
+module PlaceOS::Source
+  def self.published_status(module_id : String, control_system_id : String, status : String) : Mappings::Status
+    Mappings::Status.new(
+      status: status,
+      index: 1,
+      module_name: "Display",
+      module_id: module_id,
+      driver_id: "driver-1",
+      control_system_id: control_system_id,
+      zone_mapping: {"org" => "org-1"},
+    )
+  end
+
+  describe Published do
+    it "finds what is published by system" do
+      published = Published.new
+      a_power = published_status("mod-a", "cs-1", "power")
+      a_input = published_status("mod-a", "cs-1", "input")
+      a_other = published_status("mod-a", "cs-2", "power")
+      b_power = published_status("mod-b", "cs-2", "power")
+      [a_power, a_input, a_other, b_power].each { |event| published.add(event) }
+      # adding twice is not counted twice
+      published.add(a_power)
+
+      published.module_ids_in({"cs-1"}).should eq Set{"mod-a"}
+      published.module_ids_in({"cs-2"}).should eq Set{"mod-a", "mod-b"}
+      published.for_system("cs-1").to_set.should eq Set{a_power, a_input}
+
+      published.remove(a_power)
+      published.module_ids_in({"cs-1"}).should eq Set{"mod-a"}
+      published.remove(a_input)
+      published.module_ids_in({"cs-1"}).should be_empty
+      published.for_system("cs-1").should be_empty
+      # removing what is not there changes nothing
+      published.remove(a_input)
+      published.module_ids_in({"cs-2"}).should eq Set{"mod-a", "mod-b"}
+      published.for_module("mod-a").should eq Set{a_other}
+    end
+  end
+end
diff --git a/spec/publishing/rename_spec.cr b/spec/publishing/rename_spec.cr
index 24a37f5..3047619 100644
--- a/spec/publishing/rename_spec.cr
+++ b/spec/publishing/rename_spec.cr
@@ -11,7 +11,7 @@ module PlaceOS::Source
   def self.mapped(module_id : String, name : String, index : Int32, org : String) : Mappings
     state = Mappings::State.new
     state.drivers[module_id] = "driver-1234"
-    state.system_modules[module_id] = [{name: name, control_system_id: "cs-1", index: index}]
+    state.map_module(module_id, {name: name, control_system_id: "cs-1", index: index})
     state.system_zones["cs-1"] = {"org" => org, "building" => "b", "level" => "l", "area" => "a"}
     Mappings.new(state)
   end
diff --git a/spec/router/control_system_router_spec.cr b/spec/router/control_system_router_spec.cr
index 5c98654..9e1e635 100644
--- a/spec/router/control_system_router_spec.cr
+++ b/spec/router/control_system_router_spec.cr
@@ -87,6 +87,32 @@ module PlaceOS::Source
       mapping[second.id.as(String)][:index].should eq 2
     end
 
+    # the initial sync publishes everything, so the load reconciles nothing
+    it "reconciles a system changed after startup, not one loaded at startup" do
+      driver = Model::Generator.driver(module_name: "Display").save!
+      mod = Model::Generator.module(driver)
+      mod.custom_name = "Display"
+      mod.save!
+      other = Model::Generator.module(driver)
+      other.custom_name = "Display"
+      other.save!
+      cs = Model::Generator.control_system
+      cs.modules = [mod.id.as(String)]
+      cs.save!
+      mappings = Mappings.new
+      router = Router::ControlSystem.new(mappings, [] of PublisherManager)
+
+      router.sync(cs.id.as(String), cs)
+      mappings.system_modules_for(cs.id.as(String)).keys.should eq [mod.id]
+      mappings.reconcile_pending?.should be_false
+
+      cs.modules = [other.id.as(String), mod.id.as(String)]
+      cs.save!
+      router.sync(cs.id.as(String))
+      mappings.system_modules_for(cs.id.as(String)).keys.to_set.should eq Set{mod.id, other.id}
+      mappings.take_reconcile.should eq Set{mod.id, other.id}
+    end
+
     it "system_zones" do
       cs = Model::Generator.control_system
       cs.id = "cs-1245"
diff --git a/spec/router/module_router_spec.cr b/spec/router/module_router_spec.cr
index 826de8a..119cc43 100644
--- a/spec/router/module_router_spec.cr
+++ b/spec/router/module_router_spec.cr
@@ -6,9 +6,7 @@ module PlaceOS::Source
       # the module name is a topic segment, so a rename moves the topic
       it "updates on changes to Module name" do
         state = Mappings::State.new
-        state.system_modules["mod-renamed"] = [
-          {name: "Old Name", control_system_id: "cs-1", index: 1},
-        ]
+        state.map_module("mod-renamed", {name: "Old Name", control_system_id: "cs-1", index: 1})
         mappings = Mappings.new(state)
 
         mappings.set_system_modules("cs-1", {
@@ -23,9 +21,7 @@ module PlaceOS::Source
 
       it "updates on changes to Module index" do
         state = Mappings::State.new
-        state.system_modules["mod-reindexed"] = [
-          {name: "M'Odule", control_system_id: "cs-1", index: 1},
-        ]
+        state.map_module("mod-reindexed", {name: "M'Odule", control_system_id: "cs-1", index: 1})
         mappings = Mappings.new(state)
 
         mappings.set_system_modules("cs-1", {
@@ -41,9 +37,7 @@ module PlaceOS::Source
         state = Mappings::State.new
         state.drivers["mod-gone"] = "driver-sns"
         state.drivers["mod-stays"] = "driver-sns"
-        state.system_modules["mod-gone"] = [
-          {name: "M'Odule", control_system_id: "cs-1", index: 1},
-        ]
+        state.map_module("mod-gone", {name: "M'Odule", control_system_id: "cs-1", index: 1})
         mappings = Mappings.new(state)
 
         driver = Model::Generator.driver(module_name: "mock")
diff --git a/spec/spec_helper.cr b/spec/spec_helper.cr
index aa53be7..ec660ba 100644
--- a/spec/spec_helper.cr
+++ b/spec/spec_helper.cr
@@ -200,7 +200,7 @@ module PlaceOS::Source
     org_id = "org-donor",
   )
     state = Mappings::State.new
-    state.system_modules[module_id] = [{name: module_name, control_system_id: control_system_id, index: index}]
+    state.map_module(module_id, {name: module_name, control_system_id: control_system_id, index: index})
     state.drivers[module_id] = driver_id
     state.system_zones[control_system_id] = {
       "area"     => area_id,
diff --git a/src/source/mappings.cr b/src/source/mappings.cr
index ca0022f..7123a64 100644
--- a/src/source/mappings.cr
+++ b/src/source/mappings.cr
@@ -45,7 +45,7 @@ module PlaceOS::Source
       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) }
+        modules_in_systems(ids) | published.module_ids_in(ids)
       )
     end
 
@@ -64,10 +64,9 @@ module PlaceOS::Source
 
     # 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]) }
+        control_system_ids.each_with_object(Set(String).new) do |control_system_id, found|
+          state.system_module_ids[control_system_id]?.try { |ids| found.concat(ids) }
         end
       end
     end
@@ -218,10 +217,10 @@ module PlaceOS::Source
     def set_system_modules(control_system_id : String, system_modules : Hash(String, SystemModule))
       write do |state|
         # Clear mappings of all references to control_system_id
-        remove_system_modules(control_system_id)
+        state.unmap_system(control_system_id)
 
         system_modules.each do |module_id, new_mapping|
-          state.system_modules[module_id] << new_mapping
+          state.map_module(module_id, new_mapping)
         end
       end
     end
@@ -232,20 +231,19 @@ module PlaceOS::Source
     # 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 }
+        found = {} of String => SystemModule
+        state.system_module_ids[control_system_id]?.try &.each do |module_id|
+          state.system_modules[module_id]?.try &.each do |mapping|
+            found[module_id] = mapping if mapping[:control_system_id] == control_system_id
+          end
         end
+        found
       end
     end
 
     # Remove System Module Mappings
     def remove_system_modules(control_system_id : String)
-      # Clear mappings of all references to control_system_id
-      write do |state|
-        state.system_modules.transform_values! do |mappings|
-          mappings.reject! { |sys_mod| sys_mod[:control_system_id] == control_system_id }
-        end
-      end
+      write(&.unmap_system(control_system_id))
     end
 
     # Mappings
@@ -258,7 +256,18 @@ module PlaceOS::Source
       include JSON::Serializable
 
       # module_id => [{control_system_id: String, index: Int32}]
-      getter system_modules : Hash(String, Array(SystemModule)) = Hash(String, Array(SystemModule)).new { |h, k| h[k] = [] of SystemModule }
+      #
+      # NOTE:: change it through `map_module`, `unmap_system` and
+      # `unmap_module`, which keep `system_module_ids` in step with it
+      getter system_modules : Hash(String, Array(SystemModule)) = Hash(String, Array(SystemModule)).new
+
+      # control_system_id => the module_ids mapped in it.
+      #
+      # The reverse of `system_modules`, so working on one system does not
+      # mean scanning every Module. Systems are mapped one at a time as the
+      # routers load, so a scan per system made startup quadratic
+      @[JSON::Field(ignore: true)]
+      getter system_module_ids : Hash(String, Set(String)) = Hash(String, Set(String)).new
 
       # module_id => driver_id
       getter drivers : Hash(String, String) = {} of String => String
@@ -268,6 +277,30 @@ module PlaceOS::Source
 
       def initialize
       end
+
+      def map_module(module_id : String, mapping : SystemModule) : Nil
+        (system_modules[module_id] ||= [] of SystemModule) << mapping
+        (system_module_ids[mapping[:control_system_id]] ||= Set(String).new) << module_id
+      end
+
+      def unmap_system(control_system_id : String) : Nil
+        return unless module_ids = system_module_ids.delete(control_system_id)
+        module_ids.each do |module_id|
+          next unless list = system_modules[module_id]?
+          list.reject! { |mapping| mapping[:control_system_id] == control_system_id }
+          system_modules.delete(module_id) if list.empty?
+        end
+      end
+
+      def unmap_module(module_id : String) : Nil
+        return unless list = system_modules.delete(module_id)
+        list.each do |mapping|
+          control_system_id = mapping[:control_system_id]
+          next unless ids = system_module_ids[control_system_id]?
+          ids.delete(module_id)
+          system_module_ids.delete(control_system_id) if ids.empty?
+        end
+      end
     end
 
     private getter mappings_lock : RWLock = RWLock.new
diff --git a/src/source/published.cr b/src/source/published.cr
index c16825d..096581c 100644
--- a/src/source/published.cr
+++ b/src/source/published.cr
@@ -16,17 +16,30 @@ module PlaceOS::Source
     # module_id => statuses published for it, one per system it is mapped in
     private getter events = Hash(String, Set(Mappings::Status)).new
 
+    # control_system_id => module_id => how many of its statuses are there.
+    # So a system can be looked up without scanning everything published
+    private getter by_system = Hash(String, Hash(String, Int32)).new
+
     def add(event : Mappings::Status) : Nil
       lock.synchronize do
-        (events[event.module_id] ||= Set(Mappings::Status).new) << event
+        set = events[event.module_id] ||= Set(Mappings::Status).new
+        next if set.includes?(event)
+        set << event
+        counts = by_system[event.control_system_id] ||= Hash(String, Int32).new(0)
+        counts[event.module_id] += 1
       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?
+        next unless set = events[event.module_id]?
+        next unless set.delete(event)
+        events.delete(event.module_id) if set.empty?
+
+        next unless counts = by_system[event.control_system_id]?
+        if (counts[event.module_id] -= 1) <= 0
+          counts.delete(event.module_id)
+          by_system.delete(event.control_system_id) if counts.empty?
         end
       end
     end
@@ -35,18 +48,24 @@ module PlaceOS::Source
       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)
+    # Modules with anything published in any of these systems
+    def module_ids_in(control_system_ids : Enumerable(String)) : 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 }
+        control_system_ids.each_with_object(Set(String).new) do |control_system_id, ids|
+          by_system[control_system_id]?.try &.each_key { |module_id| ids << module_id }
         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) }
+    # Everything published in one system
+    def for_system(control_system_id : String) : Array(Mappings::Status)
+      lock.synchronize do
+        found = [] of Mappings::Status
+        by_system[control_system_id]?.try &.each_key do |module_id|
+          events[module_id]?.try &.each { |event| found << event if event.control_system_id == control_system_id }
+        end
+        found
+      end
     end
   end
 end
diff --git a/src/source/publishing/mqtt_publisher.cr b/src/source/publishing/mqtt_publisher.cr
index 87d4d21..c47e2ae 100644
--- a/src/source/publishing/mqtt_publisher.cr
+++ b/src/source/publishing/mqtt_publisher.cr
@@ -196,14 +196,15 @@ module PlaceOS::Source
       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 status topics the broker currently retains, as a client connecting
+    # now would be handed them, that `keep` returns true for. Each topic is
+    # passed to `keep` as it arrives, so only those wanted are held.
     #
     # 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)
+    def retained_state_topics(quiet : Time::Span = 2.seconds, limit : Time::Span = 2.minutes, &keep : String -> Bool) : Set(String)
       filter = File.join(MQTT_NAMESPACE, "+", "state", "#")
       topics = Set(String).new
       lock = Mutex.new
@@ -211,8 +212,9 @@ module PlaceOS::Source
 
       client.subscribe(filter) do |topic, payload, retained|
         next unless retained && !payload.empty?
+        wanted = keep.call(topic)
         lock.synchronize do
-          topics << topic
+          topics << topic if wanted
           last = Time.instant
         end
       end
@@ -285,6 +287,15 @@ module PlaceOS::Source
       )
     end
 
+    # Where the system id sits in a state topic: after the namespace, the
+    # scope, `state` and the rest of the hierarchy
+    STATE_TOPIC_SYSTEM_SEGMENT = MQTT_NAMESPACE.split('/').size + 2 + (HIERARCHY.size - 1)
+
+    # The system a state topic belongs to
+    def self.state_topic_system_id(topic : String) : String?
+      topic.split('/')[STATE_TOPIC_SYSTEM_SEGMENT]?
+    end
+
     def self.extract_zone_hierarchy(zone_mapping : Hash(String, String))
       HIERARCHY.map { |key| zone_mapping[key]? || "_" }
     end
diff --git a/src/source/router/control_system_router.cr b/src/source/router/control_system_router.cr
index 77d0cfa..244d4ab 100644
--- a/src/source/router/control_system_router.cr
+++ b/src/source/router/control_system_router.cr
@@ -35,7 +35,11 @@ module PlaceOS::Source::Router
     private getter sync_lock = Mutex.new
 
     def handle_create(control_system : PlaceOS::Model::ControlSystem) : Resource::Result
-      sync(control_system.id.as(String))
+      # While the routers load, events are only the rows just read, one per
+      # system, and change events wait until loading is done. So the row can
+      # be used as it is, rather than read again for every system
+      loaded = control_system unless startup_finished?
+      sync(control_system.id.as(String), loaded)
     end
 
     def handle_update(control_system : PlaceOS::Model::ControlSystem) : Resource::Result
@@ -53,10 +57,16 @@ module PlaceOS::Source::Router
     # 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
+    # system mapped it, and so was published nowhere.
+    #
+    # `loaded` is the system as the initial load read it. Nothing is
+    # reconciled for it: status events have not started, nothing is published
+    # yet to take away, and their initial sync publishes every Module against
+    # the mappings built here. Queueing them all would read the state of every
+    # Module a second time, in one batch
+    def sync(control_system_id : String, loaded : PlaceOS::Model::ControlSystem? = nil) : Resource::Result
       sync_lock.synchronize do
-        current = PlaceOS::Model::ControlSystem.find?(control_system_id)
+        current = loaded || 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) }
@@ -74,7 +84,7 @@ module PlaceOS::Source::Router
           mappings.remove_system_modules(control_system_id)
         end
 
-        mappings.reconcile_systems({control_system_id}) if changed
+        mappings.reconcile_systems({control_system_id}) if changed && loaded.nil?
       end
 
       Resource::Result::Success
diff --git a/src/source/router/module_router.cr b/src/source/router/module_router.cr
index 53aa8b5..d0306de 100644
--- a/src/source/router/module_router.cr
+++ b/src/source/router/module_router.cr
@@ -52,7 +52,7 @@ module PlaceOS::Source::Router
       mappings.write do |state|
         # Remove reference in drivers
         state.drivers.delete(module_id)
-        state.system_modules.delete(module_id)
+        state.unmap_module(module_id)
       end
 
       # What it published is known from the registry, not from its storage,
diff --git a/src/source/status_events.cr b/src/source/status_events.cr
index 7cfe69a..6f8d49d 100644
--- a/src/source/status_events.cr
+++ b/src/source/status_events.cr
@@ -290,27 +290,32 @@ module PlaceOS::Source
     # event: one that arrives meanwhile is processed afterwards, against the
     # new mappings, with a value at least as new as the one read here.
     #
+    # NOTE:: a zone or driver change can take in thousands of Modules. Only
+    # their status names are read up front; values are read a Module at a time
+    # as they are published, so their size never adds up in memory
+    #
     # 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}
+      # module_id => what it should now be published at, and is not
+      additions = Hash(String, Array(Mappings::Status)).new
       # 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
+        statuses = begin
+          PlaceOS::Driver::RedisStorage.new(module_id).keys
         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|
+        statuses.each do |status|
           mappings.status_events?(module_id, status).try &.each { |event| after << event }
         end
 
@@ -319,8 +324,9 @@ module PlaceOS::Source
             valid << key
           end
         end
-        (before - after).each { |event| removals << event }
-        (after - before).each { |event| additions << {event, values[event.status]} }
+        removals.concat(before - after)
+        added = after - before
+        additions[module_id] = added.to_a unless added.empty?
       end
 
       timestamp = Time.utc
@@ -331,19 +337,33 @@ module PlaceOS::Source
         message = Publisher::Message.new(event, nil, timestamp)
         publisher_managers.each(&.broadcast_delete(message))
       end
+      valid.clear
 
-      additions.each do |(event, payload)|
-        mappings.published.add(event)
-        message = Publisher::Message.new(event, payload, timestamp)
-        publisher_managers.each(&.broadcast(message))
+      added = 0
+      additions.each do |module_id, events|
+        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
+
+        events.each do |event|
+          # cleared since its name was read: its removal is on the way
+          next unless payload = values[event.status]?
+          mappings.published.add(event)
+          message = Publisher::Message.new(event, payload, timestamp)
+          publisher_managers.each(&.broadcast(message))
+          added += 1
+        end
       end
 
-      unless removals.empty? && additions.empty?
+      unless removals.empty? && added == 0
         Log.info { {
           message: "reconciled retained state",
           modules: module_ids.size.to_s,
           removed: removals.size.to_s,
-          added:   additions.size.to_s,
+          added:   added.to_s,
         } }
       end
     end
@@ -354,6 +374,11 @@ module PlaceOS::Source
     private getter sweep_lock = Mutex.new
     @pending_sweep : {MqttPublisher, Set(String)}? = nil
 
+    # Systems whose published topics are kept while a broker's retained topics
+    # stream past. The broker sends them in topic order, so each system's
+    # arrive together and a handful is plenty
+    SWEEP_SYSTEMS_CACHED = 16
+
     # Removes every status topic a broker retains that nothing maps to.
     #
     # The registry of what is published starts empty, so whatever a broker
@@ -361,6 +386,11 @@ module PlaceOS::Source
     # 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.
+    #
+    # NOTE:: a broker can retain hundreds of thousands of topics. Each is
+    # checked against the registry as it arrives and only the ones that look
+    # stale are kept, rather than holding every retained topic, and every
+    # published one, to compare as two sets
     def sweep_brokers : Nil
       # the registry is only complete once every synced event is processed
       until stopped? || (synchronize { event_container.empty? } && !mappings.reconcile_pending?)
@@ -371,7 +401,9 @@ module PlaceOS::Source
         next unless manager.is_a?(MqttBrokerManager)
         manager.publishers.each do |publisher|
           begin
-            topics = publisher.retained_state_topics
+            cache = Hash(String, Set(String)).new
+            topics = publisher.retained_state_topics { |topic| !published_topic?(topic, cache) }
+            cache.clear
             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?
@@ -392,19 +424,41 @@ module PlaceOS::Source
       end
     end
 
+    # Removes the retained topics that are still not published.
+    #
+    # They are checked again here, in the processing fiber, because a status
+    # published while the broker's topics streamed past was retained but
+    # possibly not yet in the registry when it was first checked
+    #
     # 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 = retained.reject { |topic| published_topic?(topic) }
       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,
+        message: "swept retained state",
+        broker:  label,
+        checked: retained.size.to_s,
+        removed: stale.size.to_s,
       } }
     end
 
+    # Whether something is published at this state topic, looked up through
+    # the topic's system so only that system's topics are generated. `cache`
+    # keeps the topics of recently seen systems
+    private def published_topic?(topic : String, cache : Hash(String, Set(String))? = nil) : Bool
+      return false unless control_system_id = MqttPublisher.state_topic_system_id(topic)
+
+      keys = cache.try &.[control_system_id]?
+      unless keys
+        keys = mappings.published.for_system(control_system_id).compact_map { |event| MqttPublisher.generate_key(event) }.to_set
+        if cache
+          cache.clear if cache.size >= SWEEP_SYSTEMS_CACHED
+          cache[control_system_id] = keys
+        end
+      end
+      keys.includes?(topic)
+    end
+
     def self.parse_channel(channel : String) : {String, String}
       _, module_id, status = channel.split('/')
       {module_id, status}

From 0dd33d2993fb16c4c23c3e349614054be8976e39 Mon Sep 17 00:00:00 2001
From: Stephen von Takach <steve@place.technology>
Date: Thu, 15 Oct 2026 09:12:45 +1100
Subject: [PATCH 2/3] fix: answer scope from the mappings while the routers
 load

The Module and Driver routers asked the database whether each row was in
scope as they loaded: a Module's systems and their zones, and for a Driver
every one of its Modules, then theirs. Every system and Module is already
mapped by then, so loading asks the mappings instead.

Metadata is broadcast inline rather than from a fiber per model, so the
bounded publisher queues hold loading back instead of thousands of fibers
each holding a model's JSON, and is published once per model rather than
once per top-level zone, as the topic does not depend on the zone. A system
being loaded is mapped from the zones already read with it.

With the rest of this branch and place-labs/resource#20, at 3,000 systems
and 15,000 modules the routers finish loading in 4s (26s on master) and
boot peaks at 73MiB (1.1GiB on master).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 spec/publishing/publish_metadata_spec.cr   | 13 ++++++++
 spec/router/driver_router_spec.cr          | 26 +++++++++++++++
 spec/router/module_router_spec.cr          | 19 +++++++++++
 src/source/publishing/publish_metadata.cr  | 33 ++++++++++++++-----
 src/source/router/control_system_router.cr | 17 +++++++---
 src/source/router/driver_router.cr         | 38 ++++++++++++++++++++--
 src/source/router/module_router.cr         | 18 ++++++++--
 src/source/router/zone_router.cr           |  4 +--
 8 files changed, 147 insertions(+), 21 deletions(-)

diff --git a/spec/publishing/publish_metadata_spec.cr b/spec/publishing/publish_metadata_spec.cr
index 972cb9c..6d0db2c 100644
--- a/spec/publishing/publish_metadata_spec.cr
+++ b/spec/publishing/publish_metadata_spec.cr
@@ -64,5 +64,18 @@ module PlaceOS::Source
       message.data.should eq Mappings::Metadata.new("hello", "org")
       message.payload.should eq payload.to_json
     end
+
+    # the topic is the scope, so a second top-level zone would publish it twice
+    it "publishes once however many top-level zones there are" do
+      zones = Array.new(2) do
+        zone = Model::Generator.zone
+        zone.tags = Set{Mappings.scope}
+        zone
+      end
+      router = Dummy.new
+      router.publish_metadata(zones, MockModel.new(id: "hello", some_data: "edkh"))
+
+      router.@publisher_managers.first.messages.size.should eq 1
+    end
   end
 end
diff --git a/spec/router/driver_router_spec.cr b/spec/router/driver_router_spec.cr
index 7079337..5c5f171 100644
--- a/spec/router/driver_router_spec.cr
+++ b/spec/router/driver_router_spec.cr
@@ -21,5 +21,31 @@ module PlaceOS::Source
       drivers = mappings.read(&.drivers)
       drivers.should eq({"mod-keep" => "driver-other"})
     end
+
+    # while loading, scope is read from the mappings, not the database
+    it "publishes a loaded Driver's metadata once, if it has a Module in scope" do
+      state = Mappings::State.new
+      state.drivers["mod-a"] = "driver-scoped"
+      state.drivers["mod-b"] = "driver-scoped"
+      state.drivers["mod-c"] = "driver-unscoped"
+      state.map_module("mod-a", {name: "Display", control_system_id: "cs-1", index: 1})
+      state.map_module("mod-b", {name: "Display", control_system_id: "cs-2", index: 1})
+      state.map_module("mod-c", {name: "Display", control_system_id: "cs-3", index: 1})
+      state.system_zones["cs-1"] = {Mappings.scope => "org-1", "building" => "b"}
+      state.system_zones["cs-2"] = {Mappings.scope => "org-2"}
+      state.system_zones["cs-3"] = {"building" => "b"}
+      manager = MockManager.new
+      router = Router::Driver.new(Mappings.new(state), [manager] of PublisherManager)
+
+      scoped = Model::Generator.driver(module_name: "mock")
+      scoped.id = "driver-scoped"
+      unscoped = Model::Generator.driver(module_name: "mock")
+      unscoped.id = "driver-unscoped"
+
+      router.process_resource(:created, scoped).should eq Resource::Result::Success
+      router.process_resource(:created, unscoped).should eq Resource::Result::Skipped
+
+      manager.messages.map(&.data).should eq [Mappings::Metadata.new("driver-scoped")]
+    end
   end
 end
diff --git a/spec/router/module_router_spec.cr b/spec/router/module_router_spec.cr
index 119cc43..6b986e2 100644
--- a/spec/router/module_router_spec.cr
+++ b/spec/router/module_router_spec.cr
@@ -69,6 +69,25 @@ module PlaceOS::Source
         mappings.@state.drivers["mod-1234"].should eq "driver-sns"
       end
 
+      # while loading, scope is read from the mappings, not the database
+      it "maps a loaded Module only if a system has it" do
+        driver = Model::Generator.driver(module_name: "mock")
+        driver.id = "driver-sns"
+        mapped = Model::Generator.module(driver)
+        mapped.id = "mod-mapped"
+        unmapped = Model::Generator.module(driver)
+        unmapped.id = "mod-unmapped"
+        state = Mappings::State.new
+        state.map_module("mod-mapped", {name: "Display", control_system_id: "cs-1", index: 1})
+        mappings = Mappings.new(state)
+
+        router = Router::Module.new(mappings)
+        router.process_resource(:created, mapped).should eq Resource::Result::Success
+        router.process_resource(:created, unmapped).should eq Resource::Result::Skipped
+
+        mappings.read(&.drivers).should eq({"mod-mapped" => "driver-sns"})
+      end
+
       it "removes module_id reference on destroy" do
         driver = Model::Generator.driver(module_name: "mock")
         driver.id = "driver-sns"
diff --git a/src/source/publishing/publish_metadata.cr b/src/source/publishing/publish_metadata.cr
index b6f3f35..7f6be7c 100644
--- a/src/source/publishing/publish_metadata.cr
+++ b/src/source/publishing/publish_metadata.cr
@@ -8,20 +8,35 @@ module PlaceOS::Source
   module PublishMetadata(Model)
     abstract def publisher_managers : Array(PublisherManager)
 
+    # Publish model metadata if any of `zones` is top-level.
+    #
+    # Once, however many of them are: the topic is the scope, not the zone
+    def publish_metadata(zones : Enumerable(::PlaceOS::Model::Zone), model : Model)
+      publish_metadata(model) if zones.any? { |zone| Mappings.hierarchy_tag?(zone) == Mappings.scope }
+    end
+
     # Publish model metadata
     #
     # Only models in the top-level hiearchy are published
     def publish_metadata(zone : ::PlaceOS::Model::Zone, model : Model)
-      if Mappings.hierarchy_tag?(zone) == Mappings.scope
-        message = Publisher::Message.new(
-          data: Mappings::Metadata.new(model.id.as(String)),
-          payload: model.destroyed? ? nil : model.to_json,
-          timestamp: Time.utc
-        )
+      publish_metadata({zone}, model)
+    end
 
-        # Fire off broadcast
-        spawn { publisher_managers.each &.broadcast(message) }
-      end
+    # Publish model metadata, for a model known to be in the top-level hierarchy
+    #
+    # NOTE:: broadcast here rather than from a fiber of its own. The publisher
+    # queues are bounded, and as the routers load, every system and zone is
+    # published at once: a fiber each held every one of those models' JSON in
+    # memory while it waited its turn. Waiting here holds one
+    def publish_metadata(model : Model)
+      message = Publisher::Message.new(
+        data: Mappings::Metadata.new(model.id.as(String)),
+        payload: model.destroyed? ? nil : model.to_json,
+        timestamp: Time.utc
+      )
+      publisher_managers.each &.broadcast(message)
+    rescue error
+      Log.warn(exception: error) { "publishing metadata for #{model.id}" }
     end
   end
 end
diff --git a/src/source/router/control_system_router.cr b/src/source/router/control_system_router.cr
index 244d4ab..6df1c37 100644
--- a/src/source/router/control_system_router.cr
+++ b/src/source/router/control_system_router.cr
@@ -64,11 +64,17 @@ module PlaceOS::Source::Router
     # yet to take away, and their initial sync publishes every Module against
     # the mappings built here. Queueing them all would read the state of every
     # Module a second time, in one batch
-    def sync(control_system_id : String, loaded : PlaceOS::Model::ControlSystem? = nil) : Resource::Result
+    #
+    # `loaded_zones` are the system's hierarchy zones, when already read with it
+    def sync(
+      control_system_id : String,
+      loaded : PlaceOS::Model::ControlSystem? = nil,
+      loaded_zones : Array(PlaceOS::Model::Zone)? = nil,
+    ) : Resource::Result
       sync_lock.synchronize do
         current = loaded || PlaceOS::Model::ControlSystem.find?(control_system_id)
 
-        zones = current.try { |cs| Router::ControlSystem.system_zones(cs) }
+        zones = current.try { |cs| Router::ControlSystem.system_zones(cs, loaded ? loaded_zones : nil) }
         modules = current.try { |cs| Router::ControlSystem.system_modules(cs) }
 
         changed = mappings.read do |state|
@@ -100,11 +106,14 @@ module PlaceOS::Source::Router
       # 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) }
+      publish_metadata(hierarchy_zones, control_system)
 
       case action
       in .created?
-        handle_create(control_system)
+        # while loading, the zones just read are the system's hierarchy, so
+        # they are not read again to map it
+        return handle_create(control_system) if startup_finished?
+        sync(control_system.id.as(String), control_system, hierarchy_zones)
       in .updated?
         handle_update(control_system)
       in .deleted?
diff --git a/src/source/router/driver_router.cr b/src/source/router/driver_router.cr
index b5891d7..cf89ccf 100644
--- a/src/source/router/driver_router.cr
+++ b/src/source/router/driver_router.cr
@@ -24,13 +24,18 @@ module PlaceOS::Source::Router
       driver = resource
       driver_id = driver.id.as(String)
 
+      if action.created? && !startup_finished?
+        # loading: only its metadata is published, and only if it is in scope
+        return Resource::Result::Skipped unless loaded_in_scope.includes?(driver_id)
+        publish_metadata(driver)
+        return Resource::Result::Success
+      end
+
       hierarchy_zones = Mappings.hierarchy_zones(driver)
 
       return Resource::Result::Skipped if hierarchy_zones.empty? && !action.deleted?
 
-      hierarchy_zones.each do |zone|
-        publish_metadata(zone, driver)
-      end
+      publish_metadata(hierarchy_zones, driver)
 
       if action.deleted?
         # driver_id is a topic segment, so everything published by a Module of
@@ -48,5 +53,32 @@ module PlaceOS::Source::Router
 
       Resource::Result::Success
     end
+
+    def start(timeout : Time::Span = LOAD_TIMEOUT) : self
+      @loaded_in_scope = nil
+      super
+    end
+
+    @loaded_in_scope : Set(String)? = nil
+    private getter loaded_in_scope_lock = Mutex.new
+
+    # The Drivers with a Module in a system under a top-level zone, from the
+    # mappings, which hold every system and Module by the time Drivers load.
+    #
+    # Worked out once, for loading. Asking the database meant reading every
+    # Module of each Driver, then each Module's systems and their zones
+    private def loaded_in_scope : Set(String)
+      loaded_in_scope_lock.synchronize do
+        @loaded_in_scope ||= mappings.read do |state|
+          state.drivers.each_with_object(Set(String).new) do |(module_id, driver_id), found|
+            next if found.includes?(driver_id)
+            scoped = state.system_modules[module_id]?.try &.any? do |mapping|
+              state.system_zones[mapping[:control_system_id]]?.try(&.has_key?(Mappings.scope))
+            end
+            found << driver_id if scoped
+          end
+        end
+      end
+    end
   end
 end
diff --git a/src/source/router/module_router.cr b/src/source/router/module_router.cr
index d0306de..74f0953 100644
--- a/src/source/router/module_router.cr
+++ b/src/source/router/module_router.cr
@@ -66,8 +66,7 @@ module PlaceOS::Source::Router
     def process_resource(action : Resource::Action, resource : Model::Module) : Resource::Result
       mod = resource
 
-      hierarchy_zones = Mappings.hierarchy_zones(mod)
-      return Resource::Result::Skipped if hierarchy_zones.empty? && !action.deleted?
+      return Resource::Result::Skipped unless action.deleted? || in_scope?(action, mod)
 
       case action
       in .created?
@@ -78,5 +77,20 @@ module PlaceOS::Source::Router
         handle_delete(mod)
       end
     end
+
+    # Whether the Module is in a system with a hierarchy.
+    #
+    # While loading, the systems are already mapped (the ControlSystem router
+    # loads first, and maps exactly the systems with a hierarchy), so that is
+    # asked of the mappings. Asking the database meant reading each Module's
+    # systems, and each of their zones, one Module at a time
+    private def in_scope?(action : Resource::Action, mod : Model::Module) : Bool
+      if action.created? && !startup_finished?
+        module_id = mod.id.as(String)
+        mappings.read(&.system_modules.has_key?(module_id))
+      else
+        !Mappings.hierarchy_zones(mod).empty?
+      end
+    end
   end
 end
diff --git a/src/source/router/zone_router.cr b/src/source/router/zone_router.cr
index bd0ffa6..21ff71a 100644
--- a/src/source/router/zone_router.cr
+++ b/src/source/router/zone_router.cr
@@ -37,9 +37,7 @@ module PlaceOS::Source::Router
       hierarchy_zones = Mappings.hierarchy_zones(zone)
       return Resource::Result::Skipped if hierarchy_zones.empty? && !action.deleted?
 
-      hierarchy_zones.each do |parent_zone|
-        publish_metadata(parent_zone, zone)
-      end
+      publish_metadata(hierarchy_zones, zone)
 
       # Update/destroy Zone mappings
       if zone.tags_changed? || zone.destroyed?

From 69f596c93cb2030e12e647238d30fb77451d11d9 Mon Sep 17 00:00:00 2001
From: Stephen von Takach <steve@place.technology>
Date: Thu, 15 Oct 2026 09:32:18 +1100
Subject: [PATCH 3/3] chore(shard.lock): bump placeos-resource to 3.2.0 for
 keyset loading

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 shard.lock | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/shard.lock b/shard.lock
index 1f84f7d..1095185 100644
--- a/shard.lock
+++ b/shard.lock
@@ -195,7 +195,7 @@ shards:
 
   placeos-resource:
     git: https://github.com/place-labs/resource.git
-    version: 3.1.3
+    version: 3.2.0
 
   pool:
     git: https://github.com/ysbaddaden/pool.git