feat: clear retained state when a topic moves or goes away

spec/publishing/mqtt_publisher_spec.cr

@@ -1,97 +1,6 @@
 require "../spec_helper"
 
 module PlaceOS::Source
-  # Builds a status event and its MQTT topic under a scope unique to the caller.
-  #
-  # NOTE:: the org id has to vary too, not just the module. Retained messages
-  # live in the broker until something replaces them, so a shared scope means a
-  # wildcard subscription picks up whatever earlier examples — or earlier runs
-  # against the same container — left behind
-  def self.unique_status_event(org_id : String = "org-#{UUID.random}")
-    module_id = UUID.random.to_s
-    state = mock_state(
-      module_id: module_id,
-      index: 5,
-      module_name: "M'Odule",
-      driver_id: "12345",
-      control_system_id: "cs-#{UUID.random}",
-      area_id: "2042",
-      level_id: "nek",
-      building_id: "cards",
-      org_id: org_id,
-    )
-
-    event = Mappings.new(state).status_events?(module_id, "power").not_nil!.first
-    {event, MqttPublisher.generate_key(event).not_nil!}
-  end
-
-  # A Broker of its own per example, deliberately **not** persisted.
-  #
-  # Two reasons. The publisher connects with the broker id as its MQTT client
-  # id, so examples sharing one take each other's session over — and a client
-  # that has been taken over stays down, by design.
-  #
-  # Saving it would swap that problem for a worse one: `MqttBrokerManager` is a
-  # `Resource(Model::Broker)`, so any manager still running from an earlier spec
-  # reacts to the new row, builds its own publisher for it, and takes the
-  # session over from underneath us
-  def self.isolated_broker : PlaceOS::Model::Broker
-    shared = test_broker
-    broker = PlaceOS::Model::Broker.new(
-      name: "spec-#{UUID.random}",
-      host: shared.host,
-      port: shared.port,
-      auth_type: :no_auth,
-    )
-    broker.id = "spec-broker-#{UUID.random}"
-    broker
-  end
-
-  # Runs a block with a publisher on its own broker, cleaning both up after
-  def self.with_publisher(&)
-    broker = isolated_broker
-    publisher = MqttPublisher.new(broker)
-    begin
-      yield publisher
-    ensure
-      publisher.stop rescue nil
-    end
-  end
-
-  # A bare client against the same broker, standing in for anything else that
-  # consumes the state we publish
-  def self.subscriber(broker = test_broker) : ::MQTT::Client
-    client = ::MQTT::Client.new do
-      ::MQTT::Transport::TCP.new(host: broker.host, port: broker.port).as(::MQTT::Transport)
-    end
-    client.connect(client_id: "spec-subscriber-#{UUID.random}")
-    client
-  end
-
-  # Subscribes and hands back a channel of everything that arrives
-  def self.subscribe_channel(client, topic) : Channel(Tuple(String, String, Bool))
-    received = Channel(Tuple(String, String, Bool)).new(16)
-    client.subscribe(topic) do |key, payload, retained|
-      received.send({key, String.new(payload), retained})
-      nil
-    end
-    received
-  end
-
-  def self.take(received, topic, timeout = 10.seconds)
-    select
-    when message = received.receive
-      message
-    when timeout(timeout)
-      raise "timed out waiting for a message on #{topic}"
-    end
-  end
-
-  def self.collect(client, topic, count = 1, timeout = 10.seconds)
-    received = subscribe_channel(client, topic)
-    Array(Tuple(String, String, Bool)).new(count) { take(received, topic, timeout) }
-  end
-
   describe MqttPublisher do
     describe "connection" do
       it "negotiates a protocol version with the broker" do
@@ -257,6 +166,35 @@ module PlaceOS::Source
       end
     end
 
+    # A topic only holds the current value, so when a Module moves or goes away
+    # the value it left behind is a ghost: consumers keep reading state for
+    # something that no longer publishes there
+    describe "removing retained state" do
+      it "clears the retained value for a key" do
+        status_event, key = PlaceOS::Source.unique_status_event
+
+        PlaceOS::Source.with_publisher do |publisher|
+          publisher.publish(Publisher::Message.new(status_event, "true", timestamp: Time.utc))
+          sleep 200.milliseconds
+          publisher.delete(Publisher::Message.new(status_event, nil, timestamp: Time.utc))
+          sleep 300.milliseconds
+        end
+
+        client = PlaceOS::Source.subscriber
+        begin
+          messages = PlaceOS::Source.subscribe_channel(client, key)
+          select
+          when leftover = messages.receive
+            fail "retained value should have been removed, got #{leftover.inspect}"
+          when timeout(3.seconds)
+            # nothing retained, which is the pass condition
+          end
+        ensure
+          client.disconnect rescue nil
+        end
+      end
+    end
+
     describe "keys" do
       it "creates a state event topic" do
         state = mock_state(

spec/publishing/publisher_spec.cr

@@ -0,0 +1,76 @@
+require "../spec_helper"
+
+module PlaceOS::Source
+  # Records the order work was done in, so the queue's guarantees can be
+  # asserted without a broker in the way
+  private class RecordingPublisher < Publisher
+    getter handled : Array(String) = [] of String
+
+    def publish(message : Publisher::Message)
+      handled << "publish:#{message.payload}"
+    end
+
+    def delete(message : Publisher::Message)
+      handled << "delete:#{message.payload}"
+    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
+      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.start
+      sleep 200.milliseconds
+
+      publisher.handled.first(2).should eq ["delete:deletion-0", "delete:deletion-1"]
+      publisher.handled.size.should eq 5
+      publisher.deleted.should eq 2
+      publisher.processed.should eq 3
+      publisher.stop
+    end
+
+    it "reports whether deletions are still queued" do
+      publisher = RecordingPublisher.new
+      event = Mappings::Metadata.new("mod-1234", "hello")
+
+      publisher.deletes_pending?.should be_false
+      publisher.queue_delete(Publisher::Message.new(event, "pending", Time.utc))
+      publisher.deletes_pending?.should be_true
+
+      publisher.start
+      sleep 200.milliseconds
+      publisher.deletes_pending?.should be_false
+      publisher.stop
+    end
+
+    # a consumer must notice a closed queue rather than spinning on it
+    it "stops cleanly once the queues are closed" do
+      publisher = RecordingPublisher.new
+      publisher.start
+      sleep 50.milliseconds
+
+      done = Channel(Nil).new(1)
+      spawn do
+        publisher.stop
+        done.send(nil)
+      end
+
+      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
+    end
+  end
+end

spec/publishing/rename_spec.cr

@@ -0,0 +1,112 @@
+require "../spec_helper"
+require "placeos-driver/storage"
+
+module PlaceOS::Source
+  # The module name and index are both topic segments, so changing either moves
+  # every status the Module publishes. Because everything is retained, the old
+  # topic keeps serving its last value forever unless it is cleared — a
+  # consumer would see state for a path nothing publishes to any more.
+  # Builds mappings for one module in one system, with a hierarchy it can
+  # actually resolve a topic from
+  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.system_zones["cs-1"] = {"org" => org, "building" => "b", "level" => "l", "area" => "a"}
+    Mappings.new(state)
+  end
+
+  def self.topic_for(mappings : Mappings, module_id : String, status : String) : String
+    event = mappings.status_events?(module_id, status).not_nil!.first
+    MqttPublisher.generate_key(event).not_nil!
+  end
+
+  # Publishes a status, moves the Module, and reports what the broker holds
+  # at each path afterwards
+  def self.move(index : Int32, new_name : String, new_index : Int32)
+    module_id = "mod-#{UUID.random}"
+    org = "org-#{UUID.random}"
+    status = "power"
+
+    # the Module's stored state, which is where the statuses to clear come from
+    store = PlaceOS::Driver::RedisStorage.new(module_id)
+    store[status] = "true"
+
+    mappings = mapped(module_id, "Old Name", index, org)
+    old_topic = topic_for(mappings, module_id, status)
+
+    broker = PlaceOS::Source.isolated_broker
+    publisher = MqttPublisher.new(broker)
+    manager = MockBrokerManager.new(publisher)
+    router = Router::Module.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))
+      sleep 300.milliseconds
+
+      # the rename: clear what is there, then rewrite the mapping
+      router.remove_retained_state(module_id)
+      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))
+      sleep 800.milliseconds
+
+      {old_topic, new_topic}
+    ensure
+      publisher.stop rescue nil
+      store.clear rescue nil
+    end
+  end
+
+  def self.retained_at(topic : String) : String?
+    client = PlaceOS::Source.subscriber
+    begin
+      messages = PlaceOS::Source.subscribe_channel(client, topic)
+      select
+      when message = messages.receive
+        message[1]
+      when timeout(3.seconds)
+        nil
+      end
+    ensure
+      client.disconnect rescue nil
+    end
+  end
+
+  describe "renaming a Module" do
+    it "moves the retained value to the new topic on rename" do
+      old_topic, new_topic = PlaceOS::Source.move(1, "New Name", 1)
+      old_topic.should_not eq new_topic
+
+      # the value now lives at the new path
+      payload = PlaceOS::Source.retained_at(new_topic)
+      payload.should_not be_nil
+      JSON.parse(payload.to_s)["value"].should be_true
+
+      # and nothing is left behind at the old one
+      PlaceOS::Source.retained_at(old_topic).should be_nil
+    end
+
+    it "moves the retained value to the new topic on an index change" do
+      old_topic, new_topic = PlaceOS::Source.move(1, "Old Name", 2)
+      old_topic.should_not eq new_topic
+      old_topic.should end_with "/1/power"
+      new_topic.should end_with "/2/power"
+
+      payload = PlaceOS::Source.retained_at(new_topic)
+      payload.should_not be_nil
+      JSON.parse(payload.to_s)["value"].should be_true
+
+      PlaceOS::Source.retained_at(old_topic).should be_nil
+    end
+  end
+end

spec/router/module_router_spec.cr

@@ -54,7 +54,7 @@ module PlaceOS::Source
         Router::Module.new(mappings).handle_delete(mod)
 
         mappings.read(&.drivers).should eq({"mod-stays" => "driver-sns"})
-        mappings.read { |current| current.system_modules.has_key?("mod-gone") }.should be_false
+        mappings.read(&.system_modules.has_key?("mod-gone")).should be_false
       end
     end