fix(mqtt): keep retained state topics accurate

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

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}

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

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"

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