fix(mqtt): keep startup memory and time flat on large estates

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

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

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

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

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

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"

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

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")
@@ -75,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"

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,

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